Node.js MongoDB Driver API

Source: lib/gridfs-stream/download.js

  1. var stream = require('stream'),
  2.   util = require('util');
  4. module.exports = GridFSBucketReadStream;
  6. /**
  7.  * A readable stream that enables you to read buffers from GridFS.
  8.  *
  9.  * Do not instantiate this class directly. Use `openDownloadStream()` instead.
  10.  *
  11.  * @class
  12.  * @param {Collection} chunks Handle for chunks collection
  13.  * @param {Collection} files Handle for files collection
  14.  * @param {Object} readPreference The read preference to use
  15.  * @param {Object} filter The query to use to find the file document
  16.  * @param {Object} [options=null] Optional settings.
  17.  * @param {Number} [options.sort=null] Optional sort for the file find query
  18.  * @param {Number} [options.skip=null] Optional skip for the file find query
  19.  * @param {Number} [options.start=null] Optional 0-based offset in bytes to start streaming from
  20.  * @param {Number} [options.end=null] Optional 0-based offset in bytes to stop streaming before
  21.  * @fires GridFSBucketReadStream#error
  22.  * @fires GridFSBucketReadStream#file
  23.  * @return {GridFSBucketReadStream} a GridFSBucketReadStream instance.
  24.  */
  26. function GridFSBucketReadStream(chunks, files, readPreference, filter, options) {
  27.   this.s = {
  28.     bytesRead: 0,
  29.     chunks: chunks,
  30.     cursor: null,
  31.     expected: 0,
  32.     files: files,
  33.     filter: filter,
  34.     init: false,
  35.     expectedEnd: 0,
  36.     file: null,
  37.     options: options,
  38.     readPreference: readPreference
  39.   };
  41.   stream.Readable.call(this);
  42. }
  44. util.inherits(GridFSBucketReadStream, stream.Readable);
  46. /**
  47.  * An error occurred
  48.  *
  49.  * @event GridFSBucketReadStream#error
  50.  * @type {Error}
  51.  */
  53. /**
  54.  * Fires when the stream loaded the file document corresponding to the
  55.  * provided id.
  56.  *
  57.  * @event GridFSBucketReadStream#file
  58.  * @type {object}
  59.  */
  61. /**
  62.  * Emitted when a chunk of data is available to be consumed.
  63.  *
  64.  * @event GridFSBucketReadStream#data
  65.  * @type {object}
  66.  */
  68. /**
  69.  * Fired when the stream is exhausted (no more data events).
  70.  *
  71.  * @event GridFSBucketReadStream#end
  72.  * @type {object}
  73.  */
  75. /**
  76.  * Fired when the stream is exhausted and the underlying cursor is killed
  77.  *
  78.  * @event GridFSBucketReadStream#close
  79.  * @type {object}
  80.  */
  82. /**
  83.  * Reads from the cursor and pushes to the stream.
  84.  * @method
  85.  */
  87. GridFSBucketReadStream.prototype._read = function() {
  88.   var _this = this;
  89.   if (this.destroyed) {
  90.     return;
  91.   }
  93.   waitForFile(_this, function() {
  94.     doRead(_this);
  95.   });
  96. };
  98. /**
  99.  * Sets the 0-based offset in bytes to start streaming from. Throws
  100.  * an error if this stream has entered flowing mode
  101.  * (e.g. if you've already called `on('data')`)
  102.  * @method
  103.  * @param {Number} start Offset in bytes to start reading at
  104.  * @return {GridFSBucketReadStream}
  105.  */
  107. GridFSBucketReadStream.prototype.start = function(start) {
  108.   throwIfInitialized(this);
  109.   this.s.options.start = start;
  110.   return this;
  111. };
  113. /**
  114.  * Sets the 0-based offset in bytes to start streaming from. Throws
  115.  * an error if this stream has entered flowing mode
  116.  * (e.g. if you've already called `on('data')`)
  117.  * @method
  118.  * @param {Number} end Offset in bytes to stop reading at
  119.  * @return {GridFSBucketReadStream}
  120.  */
  122. GridFSBucketReadStream.prototype.end = function(end) {
  123.   throwIfInitialized(this);
  124.   this.s.options.end = end;
  125.   return this;
  126. };
  128. /**
  129.  * Marks this stream as aborted (will never push another `data` event)
  130.  * and kills the underlying cursor. Will emit the 'end' event, and then
  131.  * the 'close' event once the cursor is successfully killed.
  132.  *
  133.  * @method
  134.  * @param {GridFSBucket~errorCallback} [callback] called when the cursor is successfully closed or an error occurred.
  135.  * @fires GridFSBucketWriteStream#close
  136.  * @fires GridFSBucketWriteStream#end
  137.  */
  139. GridFSBucketReadStream.prototype.abort = function(callback) {
  140.   var _this = this;
  141.   this.push(null);
  142.   this.destroyed = true;
  143.   if (this.s.cursor) {
  144.     this.s.cursor.close(function(error) {
  145.       _this.emit('close');
  146.       callback && callback(error);
  147.     });
  148.   } else {
  149.     if (!this.s.init) {
  150.       // If not initialized, fire close event because we will never
  151.       // get a cursor
  152.       _this.emit('close');
  153.     }
  154.     callback && callback();
  155.   }
  156. };
  158. /**
  159.  * @ignore
  160.  */
  162. function throwIfInitialized(self) {
  163.   if (self.s.init) {
  164.     throw new Error('You cannot change options after the stream has entered' +
  165.       'flowing mode!');
  166.   }
  167. }
  169. /**
  170.  * @ignore
  171.  */
  173. function doRead(_this) {
  174.   if (_this.destroyed) {
  175.     return;
  176.   }
  178.   _this.s.cursor.next(function(error, doc) {
  179.     if (_this.destroyed) {
  180.       return;
  181.     }
  182.     if (error) {
  183.       return __handleError(_this, error);
  184.     }
  185.     if (!doc) {
  186.       _this.push(null);
  187.       return _this.s.cursor.close(function(error) {
  188.         if (error) {
  189.           return __handleError(_this, error);
  190.         }
  191.         _this.emit('close');
  192.       });
  193.     }
  195.     var bytesRemaining = _this.s.file.length - _this.s.bytesRead;
  196.     var expectedN = _this.s.expected++;
  197.     var expectedLength = Math.min(_this.s.file.chunkSize,
  198.       bytesRemaining);
  200.     if (doc.n > expectedN) {
  201.       var errmsg = 'ChunkIsMissing: Got unexpected n: ' + doc.n +
  202.         ', expected: ' + expectedN;
  203.       return __handleError(_this, new Error(errmsg));
  204.     }
  206.     if (doc.n < expectedN) {
  207.       errmsg = 'ExtraChunk: Got unexpected n: ' + doc.n +
  208.         ', expected: ' + expectedN;
  209.       return __handleError(_this, new Error(errmsg));
  210.     }
  212.     var buf = Buffer.isBuffer(doc.data) ? doc.data : doc.data.buffer;
  214.     if (buf.length !== expectedLength) {
  215.       if (bytesRemaining <= 0) {
  216.         errmsg = 'ExtraChunk: Got unexpected n: ' + doc.n;
  217.         return __handleError(_this, new Error(errmsg));
  218.       }
  220.       errmsg = 'ChunkIsWrongSize: Got unexpected length: ' +
  221.         buf.length + ', expected: ' + expectedLength;
  222.       return __handleError(_this, new Error(errmsg));
  223.     }
  225.     _this.s.bytesRead += buf.length;
  227.     if (buf.length === 0) {
  228.       return _this.push(null);
  229.     }
  231.     var sliceStart = null;
  232.     var sliceEnd = null;
  234.     if (_this.s.bytesToSkip != null) {
  235.       sliceStart = _this.s.bytesToSkip;
  236.       _this.s.bytesToSkip = 0;
  237.     }
  239.     if (expectedN === _this.s.expectedEnd && _this.s.bytesToTrim != null) {
  240.       sliceEnd = _this.s.bytesToTrim;
  241.     }
  243.     // If the remaining amount of data left is < chunkSize read the right amount of data
  244.     if (_this.s.options.end && (
  245.       (_this.s.options.end - _this.s.bytesToSkip) < buf.length
  246.     )) {
  247.       sliceEnd = (_this.s.options.end - _this.s.bytesToSkip);
  248.     }
  250.     if (sliceStart != null || sliceEnd != null) {
  251.       buf = buf.slice(sliceStart || 0, sliceEnd || buf.length);
  252.     }
  254.     _this.push(buf);
  255.   })
  256. }
  258. /**
  259.  * @ignore
  260.  */
  262. function init(self) {
  263.   var findOneOptions = {};
  264.   if (self.s.readPreference) {
  265.     findOneOptions.readPreference = self.s.readPreference;
  266.   }
  267.   if (self.s.options && self.s.options.sort) {
  268.     findOneOptions.sort = self.s.options.sort;
  269.   }
  270.   if (self.s.options && self.s.options.skip) {
  271.     findOneOptions.skip = self.s.options.skip;
  272.   }
  274.   self.s.files.findOne(self.s.filter, findOneOptions, function(error, doc) {
  275.     if (error) {
  276.       return __handleError(self, error);
  277.     }
  278.     if (!doc) {
  279.       var identifier = self.s.filter._id ?
  280.         self.s.filter._id.toString() : self.s.filter.filename;
  281.       var errmsg = 'FileNotFound: file ' + identifier + ' was not found';
  282.       var err = new Error(errmsg);
  283.       err.code = 'ENOENT';
  284.       return __handleError(self, err);
  285.     }
  287.     // If document is empty, kill the stream immediately and don't
  288.     // execute any reads
  289.     if (doc.length <= 0) {
  290.       self.push(null);
  291.       return;
  292.     }
  294.     if (self.destroyed) {
  295.       // If user destroys the stream before we have a cursor, wait
  296.       // until the query is done to say we're 'closed' because we can't
  297.       // cancel a query.
  298.       self.emit('close');
  299.       return;
  300.     }
  302.     self.s.bytesToSkip = handleStartOption(self, doc, self.s.options);
  304.     var filter = { files_id: doc._id };
  306.     // Currently (MongoDB 3.4.4) skip function does not support the index,
  307.     // it needs to retrieve all the documents first and then skip them. (CS-25811)
  308.     // As work around we use $gte on the "n" field.
  309.     if (self.s.options && self.s.options.start != null){
  310.       var skip = Math.floor(self.s.options.start / doc.chunkSize);
  311.       if (skip > 0){
  312.         filter["n"] = {"$gte": skip};
  313.       }
  314.     }
  315.     self.s.cursor = self.s.chunks.find(filter).sort({ n: 1 });
  317.     if (self.s.readPreference) {
  318.       self.s.cursor.setReadPreference(self.s.readPreference);
  319.     }
  321.     self.s.expectedEnd = Math.ceil(doc.length / doc.chunkSize);
  322.     self.s.file = doc;
  323.     self.s.bytesToTrim = handleEndOption(self, doc, self.s.cursor,
  324.       self.s.options);
  325.     self.emit('file', doc);
  326.   });
  327. }
  329. /**
  330.  * @ignore
  331.  */
  333. function waitForFile(_this, callback) {
  334.   if (_this.s.file) {
  335.     return callback();
  336.   }
  338.   if (!_this.s.init) {
  339.     init(_this);
  340.     _this.s.init = true;
  341.   }
  343.   _this.once('file', function() {
  344.     callback();
  345.   })
  346. }
  348. /**
  349.  * @ignore
  350.  */
  352. function handleStartOption(stream, doc, options) {
  353.   if (options && options.start != null) {
  354.     if (options.start > doc.length) {
  355.       throw new Error('Stream start (' + options.start + ') must not be ' +
  356.         'more than the length of the file (' + doc.length +')');
  357.     }
  358.     if (options.start < 0) {
  359.       throw new Error('Stream start (' + options.start + ') must not be ' +
  360.         'negative');
  361.     }
  362.     if (options.end != null && options.end < options.start) {
  363.       throw new Error('Stream start (' + options.start + ') must not be ' +
  364.         'greater than stream end (' + options.end + ')');
  365.     }
  367.     stream.s.bytesRead = Math.floor(options.start / doc.chunkSize) *
  368.       doc.chunkSize;
  369.     stream.s.expected = Math.floor(options.start / doc.chunkSize);
  371.     return options.start - stream.s.bytesRead;
  372.   }
  373. }
  375. /**
  376.  * @ignore
  377.  */
  379. function handleEndOption(stream, doc, cursor, options) {
  380.   if (options && options.end != null) {
  381.     if (options.end > doc.length) {
  382.       throw new Error('Stream end (' + options.end + ') must not be ' +
  383.         'more than the length of the file (' + doc.length +')')
  384.     }
  385.     if (options.start < 0) {
  386.       throw new Error('Stream end (' + options.end + ') must not be ' +
  387.         'negative');
  388.     }
  390.     var start = options.start != null ?
  391.       Math.floor(options.start / doc.chunkSize) :
  392.       0;
  394.     cursor.limit(Math.ceil(options.end / doc.chunkSize) - start);
  396.     stream.s.expectedEnd = Math.ceil(options.end / doc.chunkSize);
  398.     return (Math.ceil(options.end / doc.chunkSize) * doc.chunkSize) -
  399.       options.end;
  400.   }
  401. }
  403. /**
  404.  * @ignore
  405.  */
  407. function __handleError(_this, error) {
  408.   _this.emit('error', error);
  409. }