123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317 |
- "use strict";
- Object.defineProperty(exports, "__esModule", { value: true });
- exports.GridFSBucketReadStream = void 0;
- const stream_1 = require("stream");
- const error_1 = require("../error");
- class GridFSBucketReadStream extends stream_1.Readable {
-
- constructor(chunks, files, readPreference, filter, options) {
- super();
- this.s = {
- bytesToTrim: 0,
- bytesToSkip: 0,
- bytesRead: 0,
- chunks,
- expected: 0,
- files,
- filter,
- init: false,
- expectedEnd: 0,
- options: {
- start: 0,
- end: 0,
- ...options
- },
- readPreference
- };
- }
-
- _read() {
- if (this.destroyed)
- return;
- waitForFile(this, () => doRead(this));
- }
-
- start(start = 0) {
- throwIfInitialized(this);
- this.s.options.start = start;
- return this;
- }
-
- end(end = 0) {
- throwIfInitialized(this);
- this.s.options.end = end;
- return this;
- }
-
- abort(callback) {
- this.push(null);
- this.destroyed = true;
- if (this.s.cursor) {
- this.s.cursor.close(error => {
- this.emit(GridFSBucketReadStream.CLOSE);
- callback && callback(error);
- });
- }
- else {
- if (!this.s.init) {
-
-
- this.emit(GridFSBucketReadStream.CLOSE);
- }
- callback && callback();
- }
- }
- }
- exports.GridFSBucketReadStream = GridFSBucketReadStream;
- GridFSBucketReadStream.ERROR = 'error';
- GridFSBucketReadStream.FILE = 'file';
- GridFSBucketReadStream.DATA = 'data';
- GridFSBucketReadStream.END = 'end';
- GridFSBucketReadStream.CLOSE = 'close';
- function throwIfInitialized(stream) {
- if (stream.s.init) {
- throw new error_1.MongoGridFSStreamError('Options cannot be changed after the stream is initialized');
- }
- }
- function doRead(stream) {
- if (stream.destroyed)
- return;
- if (!stream.s.cursor)
- return;
- if (!stream.s.file)
- return;
- stream.s.cursor.next((error, doc) => {
- if (stream.destroyed) {
- return;
- }
- if (error) {
- stream.emit(GridFSBucketReadStream.ERROR, error);
- return;
- }
- if (!doc) {
- stream.push(null);
- process.nextTick(() => {
- if (!stream.s.cursor)
- return;
- stream.s.cursor.close(error => {
- if (error) {
- stream.emit(GridFSBucketReadStream.ERROR, error);
- return;
- }
- stream.emit(GridFSBucketReadStream.CLOSE);
- });
- });
- return;
- }
- if (!stream.s.file)
- return;
- const bytesRemaining = stream.s.file.length - stream.s.bytesRead;
- const expectedN = stream.s.expected++;
- const expectedLength = Math.min(stream.s.file.chunkSize, bytesRemaining);
- if (doc.n > expectedN) {
- return stream.emit(GridFSBucketReadStream.ERROR, new error_1.MongoGridFSChunkError(`ChunkIsMissing: Got unexpected n: ${doc.n}, expected: ${expectedN}`));
- }
- if (doc.n < expectedN) {
- return stream.emit(GridFSBucketReadStream.ERROR, new error_1.MongoGridFSChunkError(`ExtraChunk: Got unexpected n: ${doc.n}, expected: ${expectedN}`));
- }
- let buf = Buffer.isBuffer(doc.data) ? doc.data : doc.data.buffer;
- if (buf.byteLength !== expectedLength) {
- if (bytesRemaining <= 0) {
- return stream.emit(GridFSBucketReadStream.ERROR, new error_1.MongoGridFSChunkError(`ExtraChunk: Got unexpected n: ${doc.n}, expected file length ${stream.s.file.length} bytes but already read ${stream.s.bytesRead} bytes`));
- }
- return stream.emit(GridFSBucketReadStream.ERROR, new error_1.MongoGridFSChunkError(`ChunkIsWrongSize: Got unexpected length: ${buf.byteLength}, expected: ${expectedLength}`));
- }
- stream.s.bytesRead += buf.byteLength;
- if (buf.byteLength === 0) {
- return stream.push(null);
- }
- let sliceStart = null;
- let sliceEnd = null;
- if (stream.s.bytesToSkip != null) {
- sliceStart = stream.s.bytesToSkip;
- stream.s.bytesToSkip = 0;
- }
- const atEndOfStream = expectedN === stream.s.expectedEnd - 1;
- const bytesLeftToRead = stream.s.options.end - stream.s.bytesToSkip;
- if (atEndOfStream && stream.s.bytesToTrim != null) {
- sliceEnd = stream.s.file.chunkSize - stream.s.bytesToTrim;
- }
- else if (stream.s.options.end && bytesLeftToRead < doc.data.byteLength) {
- sliceEnd = bytesLeftToRead;
- }
- if (sliceStart != null || sliceEnd != null) {
- buf = buf.slice(sliceStart || 0, sliceEnd || buf.byteLength);
- }
- stream.push(buf);
- return;
- });
- }
- function init(stream) {
- const findOneOptions = {};
- if (stream.s.readPreference) {
- findOneOptions.readPreference = stream.s.readPreference;
- }
- if (stream.s.options && stream.s.options.sort) {
- findOneOptions.sort = stream.s.options.sort;
- }
- if (stream.s.options && stream.s.options.skip) {
- findOneOptions.skip = stream.s.options.skip;
- }
- stream.s.files.findOne(stream.s.filter, findOneOptions, (error, doc) => {
- if (error) {
- return stream.emit(GridFSBucketReadStream.ERROR, error);
- }
- if (!doc) {
- const identifier = stream.s.filter._id
- ? stream.s.filter._id.toString()
- : stream.s.filter.filename;
- const errmsg = `FileNotFound: file ${identifier} was not found`;
-
- const err = new error_1.MongoRuntimeError(errmsg);
- err.code = 'ENOENT';
- return stream.emit(GridFSBucketReadStream.ERROR, err);
- }
-
-
- if (doc.length <= 0) {
- stream.push(null);
- return;
- }
- if (stream.destroyed) {
-
-
-
- stream.emit(GridFSBucketReadStream.CLOSE);
- return;
- }
- try {
- stream.s.bytesToSkip = handleStartOption(stream, doc, stream.s.options);
- }
- catch (error) {
- return stream.emit(GridFSBucketReadStream.ERROR, error);
- }
- const filter = { files_id: doc._id };
-
-
-
- if (stream.s.options && stream.s.options.start != null) {
- const skip = Math.floor(stream.s.options.start / doc.chunkSize);
- if (skip > 0) {
- filter['n'] = { $gte: skip };
- }
- }
- stream.s.cursor = stream.s.chunks.find(filter).sort({ n: 1 });
- if (stream.s.readPreference) {
- stream.s.cursor.withReadPreference(stream.s.readPreference);
- }
- stream.s.expectedEnd = Math.ceil(doc.length / doc.chunkSize);
- stream.s.file = doc;
- try {
- stream.s.bytesToTrim = handleEndOption(stream, doc, stream.s.cursor, stream.s.options);
- }
- catch (error) {
- return stream.emit(GridFSBucketReadStream.ERROR, error);
- }
- stream.emit(GridFSBucketReadStream.FILE, doc);
- return;
- });
- }
- function waitForFile(stream, callback) {
- if (stream.s.file) {
- return callback();
- }
- if (!stream.s.init) {
- init(stream);
- stream.s.init = true;
- }
- stream.once('file', () => {
- callback();
- });
- }
- function handleStartOption(stream, doc, options) {
- if (options && options.start != null) {
- if (options.start > doc.length) {
- throw new error_1.MongoInvalidArgumentError(`Stream start (${options.start}) must not be more than the length of the file (${doc.length})`);
- }
- if (options.start < 0) {
- throw new error_1.MongoInvalidArgumentError(`Stream start (${options.start}) must not be negative`);
- }
- if (options.end != null && options.end < options.start) {
- throw new error_1.MongoInvalidArgumentError(`Stream start (${options.start}) must not be greater than stream end (${options.end})`);
- }
- stream.s.bytesRead = Math.floor(options.start / doc.chunkSize) * doc.chunkSize;
- stream.s.expected = Math.floor(options.start / doc.chunkSize);
- return options.start - stream.s.bytesRead;
- }
- throw new error_1.MongoInvalidArgumentError('Start option must be defined');
- }
- function handleEndOption(stream, doc, cursor, options) {
- if (options && options.end != null) {
- if (options.end > doc.length) {
- throw new error_1.MongoInvalidArgumentError(`Stream end (${options.end}) must not be more than the length of the file (${doc.length})`);
- }
- if (options.start == null || options.start < 0) {
- throw new error_1.MongoInvalidArgumentError(`Stream end (${options.end}) must not be negative`);
- }
- const start = options.start != null ? Math.floor(options.start / doc.chunkSize) : 0;
- cursor.limit(Math.ceil(options.end / doc.chunkSize) - start);
- stream.s.expectedEnd = Math.ceil(options.end / doc.chunkSize);
- return Math.ceil(options.end / doc.chunkSize) * doc.chunkSize - options.end;
- }
- throw new error_1.MongoInvalidArgumentError('End option must be defined');
- }
|