"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.GridFSBucketWriteStream = void 0;
const stream_1 = require("stream");
const bson_1 = require("../bson");
const error_1 = require("../error");
const utils_1 = require("../utils");
const write_concern_1 = require("./../write_concern");
/**
* A writable stream that enables you to write buffers to GridFS.
*
* Do not instantiate this class directly. Use `openUploadStream()` instead.
* @public
*/
class GridFSBucketWriteStream extends stream_1.Writable {
/**
* @param bucket - Handle for this stream's corresponding bucket
* @param filename - The value of the 'filename' key in the files doc
* @param options - Optional settings.
* @internal
*/
constructor(bucket, filename, options) {
super();
options = options !== null && options !== void 0 ? options : {};
this.bucket = bucket;
this.chunks = bucket.s._chunksCollection;
this.filename = filename;
this.files = bucket.s._filesCollection;
this.options = options;
this.writeConcern = write_concern_1.WriteConcern.fromOptions(options) || bucket.s.options.writeConcern;
// Signals the write is all done
this.done = false;
this.id = options.id ? options.id : new bson_1.ObjectId();
// properly inherit the default chunksize from parent
this.chunkSizeBytes = options.chunkSizeBytes || this.bucket.s.options.chunkSizeBytes;
this.bufToStore = Buffer.alloc(this.chunkSizeBytes);
this.length = 0;
this.n = 0;
this.pos = 0;
this.state = {
streamEnd: false,
outstandingRequests: 0,
errored: false,
aborted: false
};
if (!this.bucket.s.calledOpenUploadStream) {
this.bucket.s.calledOpenUploadStream = true;
checkIndexes(this, () => {
this.bucket.s.checkedIndexes = true;
this.bucket.emit('index');
});
}
}
write(chunk, encodingOrCallback, callback) {
const encoding = typeof encodingOrCallback === 'function' ? undefined : encodingOrCallback;
callback = typeof encodingOrCallback === 'function' ? encodingOrCallback : callback;
return waitForIndexes(this, () => doWrite(this, chunk, encoding, callback));
}
abort(callback) {
return (0, utils_1.maybePromise)(callback, callback => {
if (this.state.streamEnd) {
// TODO(NODE-3485): Replace with MongoGridFSStreamClosed
return callback(new error_1.MongoAPIError('Cannot abort a stream that has already completed'));
}
if (this.state.aborted) {
// TODO(NODE-3485): Replace with MongoGridFSStreamClosed
return callback(new error_1.MongoAPIError('Cannot call abort() on a stream twice'));
}
this.state.aborted = true;
this.chunks.deleteMany({ files_id: this.id }, error => callback(error));
});
}
end(chunkOrCallback, encodingOrCallback, callback) {
const chunk = typeof chunkOrCallback === 'function' ? undefined : chunkOrCallback;
const encoding = typeof encodingOrCallback === 'function' ? undefined : encodingOrCallback;
callback =
typeof chunkOrCallback === 'function'
? chunkOrCallback
: typeof encodingOrCallback === 'function'
? encodingOrCallback
: callback;
if (this.state.streamEnd || checkAborted(this, callback))
return this;
this.state.streamEnd = true;
if (callback) {
this.once(GridFSBucketWriteStream.FINISH, (result) => {
if (callback)
callback(undefined, result);
});
}
if (!chunk) {
waitForIndexes(this, () => !!writeRemnant(this));
return this;
}
this.write(chunk, encoding, () => {
writeRemnant(this);
});
return this;
}
}
exports.GridFSBucketWriteStream = GridFSBucketWriteStream;
/** @event */
GridFSBucketWriteStream.CLOSE = 'close';
/** @event */
GridFSBucketWriteStream.ERROR = 'error';
/**
* `end()` was called and the write stream successfully wrote the file metadata and all the chunks to MongoDB.
* @event
*/
GridFSBucketWriteStream.FINISH = 'finish';
function __handleError(stream, error, callback) {
if (stream.state.errored) {
return;
}
stream.state.errored = true;
if (callback) {
return callback(error);
}
stream.emit(GridFSBucketWriteStream.ERROR, error);
}
function createChunkDoc(filesId, n, data) {
return {
_id: new bson_1.ObjectId(),
files_id: filesId,
n,
data
};
}
function checkChunksIndex(stream, callback) {
stream.chunks.listIndexes().toArray((error, indexes) => {
let index;
if (error) {
// Collection doesn't exist so create index
if (error instanceof error_1.MongoError && error.code === error_1.MONGODB_ERROR_CODES.NamespaceNotFound) {
index = { files_id: 1, n: 1 };
stream.chunks.createIndex(index, { background: false, unique: true }, error => {
if (error) {
return callback(error);
}
callback();
});
return;
}
return callback(error);
}
let hasChunksIndex = false;
if (indexes) {
indexes.forEach((index) => {
if (index.key) {
const keys = Object.keys(index.key);
if (keys.length === 2 && index.key.files_id === 1 && index.key.n === 1) {
hasChunksIndex = true;
}
}
});
}
if (hasChunksIndex) {
callback();
}
else {
index = { files_id: 1, n: 1 };
const writeConcernOptions = getWriteOptions(stream);
stream.chunks.createIndex(index, {
...writeConcernOptions,
background: true,
unique: true
}, callback);
}
});
}
function checkDone(stream, callback) {
if (stream.done)
return true;
if (stream.state.streamEnd && stream.state.outstandingRequests === 0 && !stream.state.errored) {
// Set done so we do not trigger duplicate createFilesDoc
stream.done = true;
// Create a new files doc
const filesDoc = createFilesDoc(stream.id, stream.length, stream.chunkSizeBytes, stream.filename, stream.options.contentType, stream.options.aliases, stream.options.metadata);
if (checkAborted(stream, callback)) {
return false;
}
stream.files.insertOne(filesDoc, getWriteOptions(stream), (error) => {
if (error) {
return __handleError(stream, error, callback);
}
stream.emit(GridFSBucketWriteStream.FINISH, filesDoc);
stream.emit(GridFSBucketWriteStream.CLOSE);
});
return true;
}
return false;
}
function checkIndexes(stream, callback) {
stream.files.findOne({}, { projection: { _id: 1 } }, (error, doc) => {
if (error) {
return callback(error);
}
if (doc) {
return callback();
}
stream.files.listIndexes().toArray((error, indexes) => {
let index;
if (error) {
// Collection doesn't exist so create index
if (error instanceof error_1.MongoError && error.code === error_1.MONGODB_ERROR_CODES.NamespaceNotFound) {
index = { filename: 1, uploadDate: 1 };
stream.files.createIndex(index, { background: false }, (error) => {
if (error) {
return callback(error);
}
checkChunksIndex(stream, callback);
});
return;
}
return callback(error);
}
let hasFileIndex = false;
if (indexes) {
indexes.forEach((index) => {
const keys = Object.keys(index.key);
if (keys.length === 2 && index.key.filename === 1 && index.key.uploadDate === 1) {
hasFileIndex = true;
}
});
}
if (hasFileIndex) {
checkChunksIndex(stream, callback);
}
else {
index = { filename: 1, uploadDate: 1 };
const writeConcernOptions = getWriteOptions(stream);
stream.files.createIndex(index, {
...writeConcernOptions,
background: false
}, (error) => {
if (error) {
return callback(error);
}
checkChunksIndex(stream, callback);
});
}
});
});
}
function createFilesDoc(_id, length, chunkSize, filename, contentType, aliases, metadata) {
const ret = {
_id,
length,
chunkSize,
uploadDate: new Date(),
filename
};
if (contentType) {
ret.contentType = contentType;
}
if (aliases) {
ret.aliases = aliases;
}
if (metadata) {
ret.metadata = metadata;
}
return ret;
}
function doWrite(stream, chunk, encoding, callback) {
if (checkAborted(stream, callback)) {
return false;
}
const inputBuf = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk, encoding);
stream.length += inputBuf.length;
// Input is small enough to fit in our buffer
if (stream.pos + inputBuf.length < stream.chunkSizeBytes) {
inputBuf.copy(stream.bufToStore, stream.pos);
stream.pos += inputBuf.length;
callback && callback();
// Note that we reverse the typical semantics of write's return value
// to be compatible with node's `.pipe()` function.
// True means client can keep writing.
return true;
}
// Otherwise, buffer is too big for current chunk, so we need to flush
// to MongoDB.
let inputBufRemaining = inputBuf.length;
let spaceRemaining = stream.chunkSizeBytes - stream.pos;
let numToCopy = Math.min(spaceRemaining, inputBuf.length);
let outstandingRequests = 0;
while (inputBufRemaining > 0) {
const inputBufPos = inputBuf.length - inputBufRemaining;
inputBuf.copy(stream.bufToStore, stream.pos, inputBufPos, inputBufPos + numToCopy);
stream.pos += numToCopy;
spaceRemaining -= numToCopy;
let doc;
if (spaceRemaining === 0) {
doc = createChunkDoc(stream.id, stream.n, Buffer.from(stream.bufToStore));
++stream.state.outstandingRequests;
++outstandingRequests;
if (checkAborted(stream, callback)) {
return false;
}
stream.chunks.insertOne(doc, getWriteOptions(stream), (error) => {
if (error) {
return __handleError(stream, error);
}
--stream.state.outstandingRequests;
--outstandingRequests;
if (!outstandingRequests) {
stream.emit('drain', doc);
callback && callback();
checkDone(stream);
}
});
spaceRemaining = stream.chunkSizeBytes;
stream.pos = 0;
++stream.n;
}
inputBufRemaining -= numToCopy;
numToCopy = Math.min(spaceRemaining, inputBufRemaining);
}
// Note that we reverse the typical semantics of write's return value
// to be compatible with node's `.pipe()` function.
// False means the client should wait for the 'drain' event.
return false;
}
function getWriteOptions(stream) {
const obj = {};
if (stream.writeConcern) {
obj.writeConcern = {
w: stream.writeConcern.w,
wtimeout: stream.writeConcern.wtimeout,
j: stream.writeConcern.j
};
}
return obj;
}
function waitForIndexes(stream, callback) {
if (stream.bucket.s.checkedIndexes) {
return callback(false);
}
stream.bucket.once('index', () => {
callback(true);
});
return true;
}
function writeRemnant(stream, callback) {
// Buffer is empty, so don't bother to insert
if (stream.pos === 0) {
return checkDone(stream, callback);
}
++stream.state.outstandingRequests;
// Create a new buffer to make sure the buffer isn't bigger than it needs
// to be.
const remnant = Buffer.alloc(stream.pos);
stream.bufToStore.copy(remnant, 0, 0, stream.pos);
const doc = createChunkDoc(stream.id, stream.n, remnant);
// If the stream was aborted, do not write remnant
if (checkAborted(stream, callback)) {
return false;
}
stream.chunks.insertOne(doc, getWriteOptions(stream), (error) => {
if (error) {
return __handleError(stream, error);
}
--stream.state.outstandingRequests;
checkDone(stream);
});
return true;
}
function checkAborted(stream, callback) {
if (stream.state.aborted) {
if (typeof callback === 'function') {
// TODO(NODE-3485): Replace with MongoGridFSStreamClosedError
callback(new error_1.MongoAPIError('Stream has been aborted'));
}
return true;
}
return false;
}
//# sourceMappingURL=upload.js.map