lib/reservoir-sampler.js
function reservoirStream(collection, size, opts) {
  opts = _defaults(opts || {}, {
    chunkSize: RESERVOIR_CHUNK_SIZE,
    promoteValues: true
  });
  var reservoir = new Reservoir(size);

  var stream = es.through(
    function write(data) {
      // fill reservoir with ids
      reservoir.pushSome(data);
    },
    function end() {
      // split the reservoir of ids into smaller chunks
      var chunks = _chunk(reservoir, opts.chunkSize);
      // create cursors for chunks
      var cursors = chunks.map(function(ids) {
        var cursor = collection.find({ _id: { $in: ids }}, { promoteValues: opts.promoteValues, raw: opts.raw });
        if (opts.fields) {
          cursor.project(opts.fields);
        }
        if (opts.maxTimeMS) {
          cursor.maxTimeMS(opts.maxTimeMS);
        }
        return cursor.stream();
      });
      // merge all cursors (order arbitrary) and emit docs
      es.merge(cursors).pipe(es.through(function(doc) {
        stream.emit('data', doc);
      })).on('end', function() {
        stream.emit('end');
      });
    }
  );
  return stream;
}
function reservoirStream(collection, size, opts) {
  opts = _defaults(opts || {}, {
    chunkSize: RESERVOIR_CHUNK_SIZE,
    promoteValues: true
  });
  var reservoir = new Reservoir(size);

  var stream = es.through(
    function write(data) {
      // fill reservoir with ids
      reservoir.pushSome(data);
    },
    function end() {
      // split the reservoir of ids into smaller chunks
      var chunks = _chunk(reservoir, opts.chunkSize);
      // create cursors for chunks
      var cursors = chunks.map(function(ids) {
        var cursor = collection.find({ _id: { $in: ids }}, { promoteValues: opts.promoteValues, raw: opts.raw });
        if (opts.fields) {
          cursor.project(opts.fields);
        }
        if (opts.maxTimeMS) {
          cursor.maxTimeMS(opts.maxTimeMS);
        }
        return cursor.stream();
      });
      // merge all cursors (order arbitrary) and emit docs
      es.merge(cursors).pipe(es.through(function(doc) {
        stream.emit('data', doc);
      })).on('end', function() {
        stream.emit('end');
      });
    }
  );
  return stream;
}
function reservoirStream ( collection , size , opts ) { opts = _defaults ( opts || { } , { chunkSize : RESERVOIR_CHUNK_SIZE , promoteValues : true } ) ; var reservoir = new Reservoir ( size ) ; var stream = es . through ( function write ( data ) { // fill reservoir with ids reservoir . pushSome ( data ) ; } , function end ( ) { // split the reservoir of ids into smaller chunks var chunks = _chunk ( reservoir , opts . chunkSize ) ; // create cursors for chunks var cursors = chunks . map ( function ( ids ) { var cursor = collection . find ( { _id : { $in : ids } } , { promoteValues : opts . promoteValues , raw : opts . raw } ) ; if ( opts . fields ) { cursor . project ( opts . fields ) ; } if ( opts . maxTimeMS ) { cursor . maxTimeMS ( opts . maxTimeMS ) ; } return cursor . stream ( ) ; } ) ; // merge all cursors (order arbitrary) and emit docs es . merge ( cursors ) . pipe ( es . through ( function ( doc ) { stream . emit ( 'data' , doc ) ; } ) ) . on ( 'end' , function ( ) { stream . emit ( 'end' ) ; } ) ; } ) ; return stream ; }
Does reservoir sampling and fetches the resulting documents from the
collection.

The query runs with a limit of `RESERVOIR_SAMPLE_LIMIT`. The reservoir
then samples `size` _ids from the query result. These _ids get grouped
into chunks of at most `RESERVOIR_CHUNK_SIZE`. For each chunk, a cursor
is opened to fetch the actual documents. The cursor streams are combined
and all the resulting documents are emitted downstream.

@param  {mongodb.Collection} collection The collection to sample from.
@param  {Number} size How many documents should be returned.
@param  {Object} opts
@option {Object} query to refine possible samples [default: `{}`].
@option {Number} size of the sample to capture [default: `5`].
@option {Object} fields to return for each document [default: `null`].
@option {Boolean} return document results as raw BSON buffers [default: `false`].
@option {Number} chunkSize For chunked $in queries [default: `1000`].
@return {stream.Readable}
@api private
Does reservoir sampling and fetches the resulting documents from the collection .