/**
* @fileoverview Upload manager for large file uploads
*/
import { Promise } from 'bluebird';
// -----------------------------------------------------------------------------
// Typedefs
// -----------------------------------------------------------------------------
/**
* Chunk uploaded event
* @event Chunk#uploaded
* @param {UploadPart} data The data of the uploaded chunk
* @private
*/
/**
* Chunk error event
* @event Chunk#error
* @param {Error} err The error that occurred
* @private
*/
/**
* Event for when the upload is successfully aborted
* @event ChunkedUploader#aborted
*/
/**
* Event for when the abort fails because the upload session is not destroyed.
* In general, the abort can be retried, and no new chunks will be uploaded.
* @event ChunkedUploader#abortFailed
* @param {Error} err The error that occurred
*/
/**
* Event for when a chunk fails to upload. Note that the chunk will automatically
* retry until it is successfully uploaded.
* @event ChunkedUploader#chunkError
* @param {Error} err The error that occurred during chunk upload
*/
/**
* Event for when a chunk is successfully uploaded
* @event ChunkedUploader#chunkUploaded
* @param {UploadPart} data The data for the uploaded chunk
*/
/**
* Event for when the entire upload is complete
* @event ChunkedUploader#uploadComplete
* @param {Object} file The file object for the newly-uploaded file
*/
/**
* Event for when an upload fails
* @event ChunkedUploader#error
* @param {Error} err The error that occurred
*/
type ChunkedUploaderOptions = {
retryInterval?: number;
parallelism?: number;
fileAttributes?: Record<string, any>;
};
type UploadSessionInfo = {
id: string;
part_size: number;
};
// -----------------------------------------------------------------------------
// Requirements
// -----------------------------------------------------------------------------
import { EventEmitter } from 'events';
import { Readable as ReadableStream } from 'stream';
import crypto from 'crypto';
import BoxClient from './box-client';
// -----------------------------------------------------------------------------
// Private
// -----------------------------------------------------------------------------
const DEFAULT_OPTIONS = Object.freeze({
parallelism: 4,
retryInterval: 1000,
});
/**
* Chunk of a file to be uploaded, which handles trying to upload itself until
* it succeeds.
* @private
*/
class Chunk extends EventEmitter {
client: BoxClient;
sessionID: string;
chunk: Buffer | string | null;
length: number;
offset: number;
totalSize: number;
options: ChunkedUploaderOptions;
data: any /* FIXME */;
retry: number | NodeJS.Timeout | null;
canceled: boolean;
/**
* Create a Chunk, representing a part of a file being uploaded
* @param {BoxClient} client The Box SDK client
* @param {string} sessionID The ID of the upload session the chunk belongs to
* @param {Buffer|string} chunk The chunk that was uploaded
* @param {int} offset The byte offset within the file where this chunk begins
* @param {int} totalSize The total size of the file this chunk belongs to
* @param {Object} options The options from the ChunkedUploader
* @param {int} options.retryInterval The number of ms to wait before retrying a chunk upload
*/
constructor(
client: BoxClient,
sessionID: string,
chunk: Buffer | string,
offset: number,
totalSize: number,
options: ChunkedUploaderOptions
) {
super();
this.client = client;
this.sessionID = sessionID;
this.chunk = chunk;
this.length = chunk.length;
this.offset = offset;
this.totalSize = totalSize;
this.options = options;
this.data = null;
this.retry = null;
this.canceled = false;
}
/**
* Get the final object representation of this chunk for the API
* @returns {UploadPart} The chunk object
*/
getData() {
return this.data.part;
}
/**
* Upload a chunk to the API
* @returns {void}
* @emits Chunk#uploaded
* @emits Chunk#error
*/
upload() {
this.client.files.uploadPart(
this.sessionID,
this.chunk!,
this.offset,
this.totalSize,
(err: any /* FIXME */, data: any /* FIXME */) => {
if (this.canceled) {
this.chunk = null;
return;
}
if (err) {
// handle the error or retry
if (err.statusCode) {
// an API error, probably not retryable!
this.emit('error', err);
} else {
// maybe a network error, retry
this.retry = setTimeout(
() => this.upload(),
this.options.retryInterval
);
}
return;
}
// Record the chunk data for commit, and try to free up the chunk buffer
this.data = data;
this.chunk = null;
this.emit('uploaded', data);
}
);
}
/**
* Cancel trying to upload a chunk, preventing it from retrying and clearing
* the associated buffer
* @returns {void}
*/
cancel() {
clearTimeout(this.retry as any); // number or NodeJS.Timeout
this.chunk = null;
this.canceled = true;
}
}
// -----------------------------------------------------------------------------
// Public
// -----------------------------------------------------------------------------
/** Manager for uploading a file in chunks */
class ChunkedUploader extends EventEmitter {
_client: BoxClient;
_sessionID: string;
_partSize: number;
_uploadSessionInfo: UploadSessionInfo;
_stream!: ReadableStream | null;
_streamBuffer!: Array<any>;
_file!: Buffer | string | null;
_size: number;
_options: Required<ChunkedUploaderOptions>;
_isStarted: boolean;
_numChunksInFlight: number;
_chunks: Array<any>;
_position: number;
_fileHash: crypto.Hash;
_promise?: Promise<any>;
_resolve?: Function;
_reject?: Function;
/**
* Create an upload manager
* @param {BoxClient} client The client to use to upload the file
* @param {Object} uploadSessionInfo The upload session info to use for chunked upload
* @param {ReadableStream|Buffer|string} file The file to upload
* @param {int} size The size of the file to be uploaded
* @param {Object} [options] Optional parameters
* @param {int} [options.retryInterval=1000] The number of ms to wait before retrying operations
* @param {int} [options.parallelism=4] The number of concurrent chunks to upload
* @param {Object} [options.fileAttributes] Attributes to set on the file during commit
*/
constructor(
client: BoxClient,
uploadSessionInfo: UploadSessionInfo,
file: ReadableStream | Buffer | string,
size: number,
options?: ChunkedUploaderOptions
) {
super();
this._client = client;
this._sessionID = uploadSessionInfo.id;
this._partSize = uploadSessionInfo.part_size;
this._uploadSessionInfo = uploadSessionInfo;
if (file instanceof ReadableStream) {
// Pause the stream so we can read specific chunks from it
this._stream = file.pause();
this._streamBuffer = [];
} else if (file instanceof Buffer || typeof file === 'string') {
this._file = file;
} else {
throw new TypeError('file must be a Stream, Buffer, or string!');
}
this._size = size;
this._options = Object.assign(
{},
DEFAULT_OPTIONS,
options
) as Required<ChunkedUploaderOptions>;
this._isStarted = false;
this._numChunksInFlight = 0;
this._chunks = [];
this._position = 0;
this._fileHash = crypto.createHash('sha1');
}
/**
* Start an upload
* @returns {Promise<Object>} A promise resolving to the uploaded file
*/
start() {
if (this._isStarted) {
return this._promise;
}
// Create the initial chunks
for (let i = 0; i < this._options.parallelism; i++) {
this._getNextChunk((chunk: any /* FIXME */) =>
chunk ? this._uploadChunk(chunk) : this._commit()
);
}
this._isStarted = true;
/* eslint-disable promise/avoid-new */
this._promise = new Promise((resolve, reject) => {
this._resolve = resolve;
this._reject = reject;
});
/* eslint-enable promise/avoid-new */
return this._promise;
}
/**
* Abort a running upload, which cancels all currently uploading chunks,
* attempts to free up held memory, and aborts the upload session. This
* cannot be undone or resumed.
* @returns {Promise} A promise resolving when the upload is aborted
* @emits ChunkedUploader#aborted
* @emits ChunkedUploader#abortFailed
*/
abort() {
this._chunks.forEach((chunk) => chunk.removeAllListeners().cancel());
this._chunks = [];
this._file = null;
this._stream = null;
return (
this._client.files
.abortUploadSession(this._sessionID)
/* eslint-disable promise/always-return */
.then(() => {
this.emit('aborted');
})
/* eslint-enable promise/always-return */
.catch((err: any /* FIXME */) => {
this.emit('abortFailed', err);
throw err;
})
);
}
/**
* Get the next chunk of the file to be uploaded
* @param {Function} callback Called with the next chunk of the file to be uploaded
* @returns {void}
* @private
*/
_getNextChunk(callback: Function) {
if (this._position >= this._size) {
callback(null);
return;
}
let buf;
if (this._file) {
// Buffer/string case, just get the slice we need
buf = this._file.slice(this._position, this._position + this._partSize);
} else if (this._streamBuffer.length > 0) {
buf = this._streamBuffer.shift();
} else {
// Stream case, need to read
buf = (this._stream as ReadableStream).read(this._partSize);
if (!buf) {
// stream needs to read more, retry later
setImmediate(() => this._getNextChunk(callback));
return;
} else if (buf.length > this._partSize) {
// stream is done reading and had extra data, buffer the remainder of the file
for (let i = 0; i < buf.length; i += this._partSize) {
this._streamBuffer.push(buf.slice(i, i + this._partSize));
}
buf = this._streamBuffer.shift();
}
}
this._fileHash.update(buf);
let chunk = new Chunk(
this._client,
this._sessionID,
buf,
this._position,
this._size,
this._options
);
this._position += buf.length;
callback(chunk);
}
/**
* Upload a chunk
* @param {Chunk} chunk The chunk to upload
* @returns {void}
* @emits ChunkedUploader#chunkError
* @emits ChunkedUploader#chunkUploaded
*/
_uploadChunk(chunk: any /* FIXME */) {
this._numChunksInFlight += 1;
chunk.on('error', (err: any /* FIXME */) => this.emit('chunkError', err));
chunk.on('uploaded', (data: any /* FIXME */) => {
this._numChunksInFlight -= 1;
this.emit('chunkUploaded', data);
this._getNextChunk((nextChunk: any /* FIXME */) =>
nextChunk ? this._uploadChunk(nextChunk) : this._commit()
);
});
chunk.upload();
this._chunks.push(chunk);
}
/**
* Commit the upload, finalizing it
* @returns {void}
* @emits ChunkedUploader#uploadComplete
* @emits ChunkedUploader#error
*/
_commit() {
if (!this._isStarted || this._numChunksInFlight > 0) {
return;
}
let hash = this._fileHash.digest('base64');
this._isStarted = false;
let options = Object.assign(
{
parts: this._chunks.map((c) => c.getData()),
},
this._options.fileAttributes
);
this._client.files.commitUploadSession(
this._sessionID,
hash,
options,
(err: any /* FIMXE */, file: any /* FIMXE */) => {
// It's not clear what the SDK can do here, so we just return the error and session info
// so users can retry if they wish
if (err) {
this.emit('error', {
uploadSession: this._uploadSessionInfo,
error: err,
});
this._reject!(err);
return;
}
this.emit('uploadComplete', file);
this._resolve!(file);
}
);
}
}
export = ChunkedUploader;