import { generateUuid, TransferProgressEvent } from "@azure/ms-rest-js"; import * as fs from "fs"; import { Readable } from "stream"; import { Aborter } from "./Aborter"; import { BlobURL } from "./BlobURL"; import { BlockBlobURL } from "./BlockBlobURL"; import { BlobHTTPHeaders } from "./generated/src/models"; import { BlobUploadCommonResponse, IDownloadFromBlobOptions, IUploadToBlockBlobOptions } from "./highlevel.common"; import { IBlobAccessConditions } from "./models"; import { Batch } from "./utils/Batch"; import { BufferScheduler } from "./utils/BufferScheduler"; import { BLOCK_BLOB_MAX_BLOCKS, BLOCK_BLOB_MAX_STAGE_BLOCK_BYTES, BLOCK_BLOB_MAX_UPLOAD_BLOB_BYTES, DEFAULT_BLOB_DOWNLOAD_BLOCK_BYTES, } from "./utils/constants"; import { generateBlockID } from "./utils/utils.common"; import { streamToBuffer } from "./utils/utils.node"; /** * ONLY AVAILABLE IN NODE.JS RUNTIME. * * Uploads a local file in blocks to a block blob. * * When file size <= 256MB, this method will use 1 upload call to finish the upload. * Otherwise, this method will call stageBlock to upload blocks, and finally call commitBlockList * to commit the block list. * * @export * @param {Aborter} aborter Create a new Aborter instance with Aborter.none or Aborter.timeout(), * goto documents of Aborter for more examples about request cancellation * @param {string} filePath Full path of local file * @param {BlockBlobURL} blockBlobURL BlockBlobURL * @param {IUploadToBlockBlobOptions} [options] IUploadToBlockBlobOptions * @returns {(Promise)} ICommonResponse */ export async function uploadFileToBlockBlob( aborter: Aborter, filePath: string, blockBlobURL: BlockBlobURL, options?: IUploadToBlockBlobOptions ): Promise { const size = fs.statSync(filePath).size; return uploadResetableStreamToBlockBlob( aborter, (offset, count) => fs.createReadStream(filePath, { autoClose: true, end: count ? offset + count - 1 : Infinity, start: offset }), size, blockBlobURL, options ); } /** * ONLY AVAILABLE IN NODE.JS RUNTIME. * * Accepts a Node.js Readable stream factory, and uploads in blocks to a block blob. * The Readable stream factory must returns a Node.js Readable stream starting from the offset defined. The offset * is the offset in the block blob to be uploaded. * * When buffer length <= 256MB, this method will use 1 upload call to finish the upload. * Otherwise, this method will call stageBlock to upload blocks, and finally call commitBlockList * to commit the block list. * * @export * @param {Aborter} aborter Create a new Aborter instance with Aborter.none or Aborter.timeout(), * goto documents of Aborter for more examples about request cancellation * @param {(offset: number) => NodeJS.ReadableStream} streamFactory Returns a Node.js Readable stream starting * from the offset defined * @param {number} size Size of the block blob * @param {BlockBlobURL} blockBlobURL BlockBlobURL * @param {IUploadToBlockBlobOptions} [options] IUploadToBlockBlobOptions * @returns {(Promise)} ICommonResponse */ async function uploadResetableStreamToBlockBlob( aborter: Aborter, streamFactory: (offset: number, count?: number) => NodeJS.ReadableStream, size: number, blockBlobURL: BlockBlobURL, options: IUploadToBlockBlobOptions = {} ): Promise { if (!options.blockSize) { options.blockSize = 0; } if (options.blockSize < 0 || options.blockSize > BLOCK_BLOB_MAX_STAGE_BLOCK_BYTES) { throw new RangeError( `blockSize option must be >= 0 and <= ${BLOCK_BLOB_MAX_STAGE_BLOCK_BYTES}` ); } if (options.maxSingleShotSize !== 0 && !options.maxSingleShotSize) { options.maxSingleShotSize = BLOCK_BLOB_MAX_UPLOAD_BLOB_BYTES; } if ( options.maxSingleShotSize < 0 || options.maxSingleShotSize > BLOCK_BLOB_MAX_UPLOAD_BLOB_BYTES ) { throw new RangeError( `maxSingleShotSize option must be >= 0 and <= ${BLOCK_BLOB_MAX_UPLOAD_BLOB_BYTES}` ); } if (options.blockSize === 0) { if (size > BLOCK_BLOB_MAX_BLOCKS * BLOCK_BLOB_MAX_STAGE_BLOCK_BYTES) { throw new RangeError(`${size} is too larger to upload to a block blob.`); } if (size > options.maxSingleShotSize) { options.blockSize = Math.ceil(size / BLOCK_BLOB_MAX_BLOCKS); if (options.blockSize < DEFAULT_BLOB_DOWNLOAD_BLOCK_BYTES) { options.blockSize = DEFAULT_BLOB_DOWNLOAD_BLOCK_BYTES; } } } if (!options.blobHTTPHeaders) { options.blobHTTPHeaders = {}; } if (!options.blobAccessConditions) { options.blobAccessConditions = {}; } if (size <= options.maxSingleShotSize) { return blockBlobURL.upload(aborter, () => streamFactory(0), size, options); } const numBlocks: number = Math.floor((size - 1) / options.blockSize) + 1; if (numBlocks > BLOCK_BLOB_MAX_BLOCKS) { throw new RangeError( `The buffer's size is too big or the BlockSize is too small;` + `the number of blocks must be <= ${BLOCK_BLOB_MAX_BLOCKS}` ); } const blockList: string[] = []; const blockIDPrefix = generateUuid(); let transferProgress: number = 0; const batch = new Batch(options.parallelism); for (let i = 0; i < numBlocks; i++) { batch.addOperation( async (): Promise => { const blockID = generateBlockID(blockIDPrefix, i); const start = options.blockSize! * i; const end = i === numBlocks - 1 ? size : start + options.blockSize!; const contentLength = end - start; blockList.push(blockID); await blockBlobURL.stageBlock( aborter, blockID, () => streamFactory(start, contentLength), contentLength, { leaseAccessConditions: options.blobAccessConditions!.leaseAccessConditions } ); // Update progress after block is successfully uploaded to server, in case of block trying transferProgress += contentLength; if (options.progress) { options.progress({ loadedBytes: transferProgress }); } } ); } await batch.do(); return blockBlobURL.commitBlockList(aborter, blockList, options); } /** * ONLY AVAILABLE IN NODE.JS RUNTIME. * * Downloads an Azure Blob in parallel to a buffer. * Offset and count are optional, pass 0 for both to download the entire blob. * * @export * @param {Aborter} aborter Create a new Aborter instance with Aborter.none or Aborter.timeout(), * goto documents of Aborter for more examples about request cancellation * @param {Buffer} buffer Buffer to be fill, must have length larger than count * @param {BlobURL} blobURL A BlobURL object * @param {number} offset From which position of the block blob to download, in bytes * @param {number} [count] How much data in bytes to be downloaded. Will download to the end when passing undefined * @param {IDownloadFromBlobOptions} [options] IDownloadFromBlobOptions * @returns {Promise} */ export async function downloadBlobToBuffer( aborter: Aborter, buffer: Buffer, blobURL: BlobURL, offset: number, count?: number, options: IDownloadFromBlobOptions = {} ): Promise { if (!options.blockSize) { options.blockSize = 0; } if (options.blockSize < 0) { throw new RangeError("blockSize option must be >= 0"); } if (options.blockSize === 0) { options.blockSize = DEFAULT_BLOB_DOWNLOAD_BLOCK_BYTES; } if (offset < 0) { throw new RangeError("offset option must be >= 0"); } if (count && count <= 0) { throw new RangeError("count option must be > 0"); } if (!options.blobAccessConditions) { options.blobAccessConditions = {}; } // Customer doesn't specify length, get it if (!count) { const response = await blobURL.getProperties(aborter, options); count = response.contentLength! - offset; if (count < 0) { throw new RangeError( `offset ${offset} shouldn't be larger than blob size ${response.contentLength!}` ); } } if (buffer.length < count) { throw new RangeError( `The buffer's size should be equal to or larger than the request count of bytes: ${count}` ); } let transferProgress: number = 0; const batch = new Batch(options.parallelism); for (let off = offset; off < offset + count; off = off + options.blockSize) { batch.addOperation(async () => { // Exclusive chunk end position let chunkEnd = offset + count!; if (off + options.blockSize! < chunkEnd) { chunkEnd = off + options.blockSize!; } const response = await blobURL.download(aborter, off, chunkEnd - off, { blobAccessConditions: options.blobAccessConditions, maxRetryRequests: options.maxRetryRequestsPerBlock }); const stream = response.readableStreamBody!; await streamToBuffer(stream, buffer, off - offset, chunkEnd - offset); // Update progress after block is downloaded, in case of block trying // Could provide finer grained progress updating inside HTTP requests, // only if convenience layer download try is enabled transferProgress += chunkEnd - off; if (options.progress) { options.progress({ loadedBytes: transferProgress }); } }); } await batch.do(); } /** * Option interface for uploadStreamToBlockBlob. * * @export * @interface IUploadStreamToBlockBlobOptions */ export interface IUploadStreamToBlockBlobOptions { /** * Blob HTTP Headers. * * @type {BlobHTTPHeaders} * @memberof IUploadStreamToBlockBlobOptions */ blobHTTPHeaders?: BlobHTTPHeaders; /** * Metadata of block blob. * * @type {{ [propertyName: string]: string }} * @memberof IUploadStreamToBlockBlobOptions */ metadata?: { [propertyName: string]: string }; /** * Access conditions headers. * * @type {IBlobAccessConditions} * @memberof IUploadStreamToBlockBlobOptions */ accessConditions?: IBlobAccessConditions; /** * Progress updater. * * @memberof IUploadStreamToBlockBlobOptions */ progress?: (progress: TransferProgressEvent) => void; } /** * ONLY AVAILABLE IN NODE.JS RUNTIME. * * Uploads a Node.js Readable stream into block blob. * * PERFORMANCE IMPROVEMENT TIPS: * * Input stream highWaterMark is better to set a same value with bufferSize * parameter, which will avoid Buffer.concat() operations. * * @export * @param {Aborter} aborter Create a new Aborter instance with Aborter.none or Aborter.timeout(), * goto documents of Aborter for more examples about request cancellation * @param {Readable} stream Node.js Readable stream * @param {BlockBlobURL} blockBlobURL A BlockBlobURL instance * @param {number} bufferSize Size of every buffer allocated, also the block size in the uploaded block blob * @param {number} maxBuffers Max buffers will allocate during uploading, positive correlation * with max uploading concurrency * @param {IUploadStreamToBlockBlobOptions} [options] * @returns {Promise} */ export async function uploadStreamToBlockBlob( aborter: Aborter, stream: Readable, blockBlobURL: BlockBlobURL, bufferSize: number, maxBuffers: number, options: IUploadStreamToBlockBlobOptions = {} ): Promise { if (!options.blobHTTPHeaders) { options.blobHTTPHeaders = {}; } if (!options.accessConditions) { options.accessConditions = {}; } let blockNum = 0; const blockIDPrefix = generateUuid(); let transferProgress: number = 0; const blockList: string[] = []; const scheduler = new BufferScheduler( stream, bufferSize, maxBuffers, async (buffer: Buffer) => { const blockID = generateBlockID(blockIDPrefix, blockNum); blockList.push(blockID); blockNum++; await blockBlobURL.stageBlock(aborter, blockID, buffer, buffer.length, { leaseAccessConditions: options.accessConditions!.leaseAccessConditions }); // Update progress after block is successfully uploaded to server, in case of block trying transferProgress += buffer.length; if (options.progress) { options.progress({ loadedBytes: transferProgress }); } }, // Parallelism should set a smaller value than maxBuffers, which is helpful to // reduce the possibility when a outgoing handler waits for stream data, in // this situation, outgoing handlers are blocked. // Outgoing queue shouldn't be empty. Math.ceil((maxBuffers / 4) * 3) ); await scheduler.do(); return blockBlobURL.commitBlockList(aborter, blockList, options); }