feat(Cursor): adds support for AsyncIterator in cursors

Adds support for async iterators in cursors, allowing
users to use for..await..of blocks on cursors.
Note: this is only loaded if Symbol.asyncIterator is supported
(v10.x).

Fixes NODE-1684

test: adds single file responsible for loading esnext tests
Now use a single file to load tests that cannot run in current env

Author: daprahamian

lib/aggregation_cursor.js

@@ -130,6 +130,11 @@ inherits(AggregationCursor, Readable);
 for (var name in CoreCursor.prototype) {
   AggregationCursor.prototype[name] = CoreCursor.prototype[name];
 }
+if (Symbol.asyncIterator) {
+  AggregationCursor.prototype[
+    Symbol.asyncIterator
+  ] = require('./async/async_iterator').asyncIterator;
+}
 
 /**
  * Set the batch size for the cursor.

lib/async/.eslintrc

@@ -0,0 +1,5 @@
+{
+  "parserOptions": {
+    "ecmaVersion": 2018
+  }
+}

lib/async/async_iterator.js

@@ -0,0 +1,15 @@
+'use strict';
+
+async function* asyncIterator() {
+  while (true) {
+    const value = await this.next();
+    if (!value) {
+      await this.close();
+      return;
+    }
+
+    yield value;
+  }
+}
+
+exports.asyncIterator = asyncIterator;

lib/command_cursor.js

@@ -156,6 +156,10 @@ for (var i = 0; i < methodsToInherit.length; i++) {
   CommandCursor.prototype[methodsToInherit[i]] = CoreCursor.prototype[methodsToInherit[i]];
 }
 
+if (Symbol.asyncIterator) {
+  CommandCursor.prototype[Symbol.asyncIterator] = require('./async/async_iterator').asyncIterator;
+}
+
 /**
  * Set the ReadPreference for the cursor.
  * @method

lib/cursor.js

@@ -203,6 +203,10 @@ function Cursor(bson, ns, cmd, options, topology, topologyOptions) {
 // Inherit from Readable
 inherits(Cursor, Readable);
 
+if (Symbol.asyncIterator) {
+  Cursor.prototype[Symbol.asyncIterator] = require('./async/async_iterator').asyncIterator;
+}
+
 // Map core cursor _next method so we can apply mapping
 Cursor.prototype._next = function() {
   if (this._initImplicitSession) {

test/functional/load_esnext.js

@@ -0,0 +1,39 @@
+'use strict';
+
+function loadTests() {
+  const fs = require('fs');
+  const path = require('path');
+
+  const directory = path.resolve.apply(path, arguments);
+  fs
+    .readdirSync(directory)
+    .filter(filePath => filePath.match(/.*\.js$/))
+    .map(filePath => path.resolve(directory, filePath))
+    .forEach(x => require(x));
+}
+
+describe('ES2017', function() {
+  let supportES2017 = false;
+  try {
+    new Function('return (async function foo() {})();')();
+    supportES2017 = true;
+  } catch (e) {
+    supportES2017 = false;
+  }
+
+  if (supportES2017) {
+    loadTests(__dirname, '..', 'examples');
+  } else {
+    it.skip('skipping ES2017 tests due to insufficient node version', function() {});
+  }
+});
+
+describe('ES2018', function() {
+  const supportES2018 = !!Symbol.asyncIterator;
+
+  if (supportES2018) {
+    loadTests(__dirname, '..', 'node-next', 'es2018');
+  } else {
+    it.skip('skipping ES2018 tests due to insufficient node version', function() {});
+  }
+});

test/functional/load_examples.js

@@ -0,0 +1,5 @@
+{
+  "parserOptions": {
+    "ecmaVersion": 2018
+  }
+}

test/node-next/es2018/.eslintrc

@@ -0,0 +1,92 @@
+'use strict';
+
+const { expect } = require('chai');
+const { MongoError } = require('../../../index');
+
+describe('Cursor Async Iterator Tests', function() {
+  let client, collection;
+  before(async function() {
+    client = this.configuration.newClient();
+
+    await client.connect();
+    const docs = Array.from({ length: 1000 }).map((_, index) => ({ foo: index, bar: 1 }));
+
+    collection = client.db(this.configuration.db).collection('async_cursor_tests');
+
+    await collection.deleteMany({});
+    await collection.insertMany(docs);
+    await client.close();
+  });
+
+  beforeEach(async function() {
+    client = this.configuration.newClient();
+    await client.connect();
+    collection = client.db(this.configuration.db).collection('async_cursor_tests');
+  });
+
+  afterEach(() => client.close());
+
+  it('should be able to use a for-await loop on a find command cursor', {
+    metadata: { requires: { node: '>=10.5.0' } },
+    test: async function() {
+      const cursor = collection.find({ bar: 1 });
+
+      let counter = 0;
+      for await (const doc of cursor) {
+        expect(doc).to.have.property('bar', 1);
+        counter += 1;
+      }
+
+      expect(counter).to.equal(1000);
+    }
+  });
+
+  it('should be able to use a for-await loop on an aggregation cursor', {
+    metadata: { requires: { node: '>=10.5.0' } },
+    test: async function() {
+      const cursor = collection.aggregate([{ $match: { bar: 1 } }]);
+
+      let counter = 0;
+      for await (const doc of cursor) {
+        expect(doc).to.have.property('bar', 1);
+        counter += 1;
+      }
+
+      expect(counter).to.equal(1000);
+    }
+  });
+
+  it('should be able to use a for-await loop on a command cursor', {
+    metadata: { requires: { node: '>=10.5.0', mongodb: '>=3.0.0' } },
+    test: async function() {
+      const cursor1 = collection.listIndexes();
+      const cursor2 = collection.listIndexes();
+
+      const indexes = await cursor1.toArray();
+      let counter = 0;
+      for await (const doc of cursor2) {
+        expect(doc).to.exist;
+        counter += 1;
+      }
+
+      expect(counter).to.equal(indexes.length);
+    }
+  });
+
+  it('should properly error when cursor is closed', {
+    metadata: { requires: { node: '>=10.5.0' } },
+    test: async function() {
+      const cursor = collection.find();
+
+      try {
+        for await (const doc of cursor) {
+          expect(doc).to.exist;
+          cursor.close();
+        }
+        throw new Error('expected closing the cursor to break iteration');
+      } catch (e) {
+        expect(e).to.be.an.instanceOf(MongoError);
+      }
+    }
+  });
+});