Cleanup

node_modules/mongoose/lib/helpers/cursor/eachAsync.js

@@ -4,6 +4,7 @@
  * Module dependencies.
  */
 
+const immediate = require('../immediate');
 const promiseOrCallback = require('../promiseOrCallback');
 
 /**
@@ -22,9 +23,22 @@ const promiseOrCallback = require('../promiseOrCallback');
 
 module.exports = function eachAsync(next, fn, options, callback) {
   const parallel = options.parallel || 1;
+  const batchSize = options.batchSize;
   const enqueue = asyncQueue();
 
   return promiseOrCallback(callback, cb => {
+    if (batchSize != null) {
+      if (typeof batchSize !== 'number') {
+        throw new TypeError('batchSize must be a number');
+      }
+      if (batchSize < 1) {
+        throw new TypeError('batchSize must be at least 1');
+      }
+      if (batchSize !== Math.floor(batchSize)) {
+        throw new TypeError('batchSize must be a positive integer');
+      }
+    }
+
     iterate(cb);
   });
 
@@ -32,6 +46,7 @@ module.exports = function eachAsync(next, fn, options, callback) {
     let drained = false;
     let handleResultsInProgress = 0;
     let currentDocumentIndex = 0;
+    let documentsBatch = [];
 
     let error = null;
     for (let i = 0; i < parallel; ++i) {
@@ -56,6 +71,8 @@ module.exports = function eachAsync(next, fn, options, callback) {
           drained = true;
           if (handleResultsInProgress <= 0) {
             finalCallback(null);
+          } else if (batchSize != null && documentsBatch.length) {
+            handleNextResult(documentsBatch, currentDocumentIndex++, handleNextResultCallBack);
           }
           return done();
         }
@@ -64,10 +81,27 @@ module.exports = function eachAsync(next, fn, options, callback) {
 
         // Kick off the subsequent `next()` before handling the result, but
         // make sure we know that we still have a result to handle re: #8422
-        process.nextTick(() => done());
+        immediate(() => done());
 
-        handleNextResult(doc, currentDocumentIndex++, function(err) {
-          --handleResultsInProgress;
+        if (batchSize != null) {
+          documentsBatch.push(doc);
+        }
+
+        // If the current documents size is less than the provided patch size don't process the documents yet
+        if (batchSize != null && documentsBatch.length !== batchSize) {
+          setTimeout(() => enqueue(fetch), 0);
+          return;
+        }
+
+        const docsToProcess = batchSize != null ? documentsBatch : doc;
+
+        function handleNextResultCallBack(err) {
+          if (batchSize != null) {
+            handleResultsInProgress -= documentsBatch.length;
+            documentsBatch = [];
+          } else {
+            --handleResultsInProgress;
+          }
           if (err != null) {
             error = err;
             return finalCallback(err);
@@ -77,7 +111,9 @@ module.exports = function eachAsync(next, fn, options, callback) {
           }
 
           setTimeout(() => enqueue(fetch), 0);
-        });
+        }
+
+        handleNextResult(docsToProcess, currentDocumentIndex++, handleNextResultCallBack);
       });
     }
   }