feat(cursor): WIP continueOnError support for eachAsync() re: #6355

lib/cursor/QueryCursor.js

@@ -225,6 +225,8 @@ QueryCursor.prototype.next = function(callback) {
  * @param {Function} fn
  * @param {Object} [options]
  * @param {Number} [options.parallel] the number of promises to execute in parallel. Defaults to 1.
+ * @param {Number} [options.batchSize] if set, will call `fn()` with arrays of documents with length at most `batchSize`
+ * @param {Boolean} [options.continueOnError=false] if true, `eachAsync()` iterates through all docs even if `fn` throws an error. If false, `eachAsync()` throws an error immediately if the given function `fn()` throws an error.
  * @param {Function} [callback] executed when all docs have been processed
  * @return {Promise}
  * @api public

lib/error/eachAsyncMultiError.js

@@ -0,0 +1,38 @@
+/*!
+ * Module dependencies.
+ */
+
+'use strict';
+
+const MongooseError = require('./');
+
+
+/**
+ * The connection failed to reconnect and will never successfully reconnect to
+ * MongoDB without manual intervention.
+ * @api private
+ */
+class EachAsyncMultiError extends MongooseError {
+  /**
+   * @param {String} connectionString
+   */
+  constructor(errors) {
+    let preview = errors.map(e => e.message).join(', ');
+    if (preview.length > 50) {
+      preview = preview.slice(0, 50) + '...';
+    }
+    super(`eachAsync() finished with ${errors.length} errors: ${preview}`);
+
+    this.errors = errors;
+  }
+}
+
+Object.defineProperty(EachAsyncMultiError.prototype, 'name', {
+  value: 'EachAsyncMultiError'
+});
+
+/*!
+ * exports
+ */
+
+module.exports = EachAsyncMultiError;

lib/helpers/cursor/eachAsync.js

@@ -4,6 +4,7 @@
  * Module dependencies.
  */
 
+const EachAsyncMultiError = require('../../error/eachAsyncMultiError');
 const immediate = require('../immediate');
 const promiseOrCallback = require('../promiseOrCallback');
 
@@ -24,10 +25,11 @@ const promiseOrCallback = require('../promiseOrCallback');
 module.exports = function eachAsync(next, fn, options, callback) {
   const parallel = options.parallel || 1;
   const batchSize = options.batchSize;
+  const continueOnError = options.continueOnError;
+  const errors = [];
   const enqueue = asyncQueue();
 
   return promiseOrCallback(callback, cb => {
-
     if (batchSize != null) {
       if (typeof batchSize !== 'number') {
         throw new TypeError('batchSize must be a number');
@@ -62,6 +64,7 @@ module.exports = function eachAsync(next, fn, options, callback) {
           return done();
         }
         if (err != null) {
+          errors.push(err);
           error = err;
           finalCallback(err);
           return done();

types/cursor.d.ts

@@ -3,6 +3,12 @@ import stream = require('stream');
 declare module 'mongoose' {
   type CursorFlag = 'tailable' | 'oplogReplay' | 'noCursorTimeout' | 'awaitData' | 'partial';
 
+  interface EachAsyncOptions {
+    parallel?: number;
+    batchSize?: number;
+    continueOnError?: boolean;
+  }
+
   class Cursor<DocType = any, Options = never> extends stream.Readable {
     [Symbol.asyncIterator](): AsyncIterableIterator<DocType>;
 
@@ -25,10 +31,10 @@ declare module 'mongoose' {
      * will wait for the promise to resolve before iterating on to the next one.
      * Returns a promise that resolves when done.
      */
-    eachAsync(fn: (doc: DocType[]) => any, options: { parallel?: number, batchSize: number }, callback: CallbackWithoutResult): void;
-    eachAsync(fn: (doc: DocType) => any, options: { parallel?: number }, callback: CallbackWithoutResult): void;
-    eachAsync(fn: (doc: DocType[]) => any, options: { parallel?: number, batchSize: number }): Promise<void>;
-    eachAsync(fn: (doc: DocType) => any, options?: { parallel?: number }): Promise<void>;
+    eachAsync(fn: (doc: DocType[]) => any, options: EachAsyncOptions, callback: CallbackWithoutResult): void;
+    eachAsync(fn: (doc: DocType) => any, options: EachAsyncOptions, callback: CallbackWithoutResult): void;
+    eachAsync(fn: (doc: DocType[]) => any, options: EachAsyncOptions): Promise<void>;
+    eachAsync(fn: (doc: DocType) => any, options?: EachAsyncOptions): Promise<void>;
 
     /**
      * Registers a transform function which subsequently maps documents retrieved