Merge pull request #1491 from mbroadst/NODE-953

fix(collection): pass batchSize to AggregationCursor

Author: christkv

lib/collection.js

@@ -21,7 +21,8 @@ var checkCollectionName = require('./utils').checkCollectionName
   , Cursor = require('./cursor')
   , unordered = require('./bulk/unordered')
   , ordered = require('./bulk/ordered')
-  , assign = require('./utils').assign;
+  , assign = require('./utils').assign
+  , mergeOptions = require('./utils').mergeOptions;
 
 /**
  * @fileOverview The **Collection** class is an internal class that embodies a MongoDB collection
@@ -2668,6 +2669,7 @@ Collection.prototype.aggregate = function(pipeline, options, callback) {
     if(this.s.topology.capabilities().hasAggregationCursor) {
       options.cursor = options.cursor || { batchSize : 1000 };
       command.cursor = options.cursor;
+      mergeOptions(options, options.cursor);
     }
 
     // Allow disk usage command

test/functional/aggregation_tests.js

@@ -827,3 +827,42 @@ exports['Should correctly handle ISODate date matches in aggregation framework']
     // DOC_END
   }
 }
+
+exports['should maintain batch size between calls to receive new batches'] = {
+  metadata: {
+    requires: { generators: true, topology: 'single' , node: ">0.10.0" }
+  },
+
+  // The actual test we wish to run
+  test: function(configure, test) {
+    var co = require('co');
+
+    co(function*() {
+      var instance = configure.newDbInstance({ w: 1 }, { poolSize: 1 });
+      var db = yield instance.open();
+
+      var docs = [ { a: 1 }, { a: 1 }, { a: 1 }, { a: 1 }, { a: 1 }, { a: 1 } ];
+      var collection = db.collection('batchSizeContinue');
+      yield collection.insertMany(docs, { w: 1 });
+      var cursor = collection.aggregate([
+          { $match: { a: 1 } }, { $limit: 6 }
+        ], {
+          cursor: { batchSize: 2 }
+        });
+
+      var count = 0;
+      while (yield cursor.hasNext()) {
+        var data = yield cursor.next();
+        test.equal(data.a, 1);
+
+        // ensure batch size is as specified
+        test.equal(cursor.cursorState.documents.length, 2);
+        count++;
+      }
+
+      test.equal(count, 6);
+      db.close();
+      test.done();
+    });
+  }
+}