[wip] initial async generator support

8 files changed, 168 insertions, 6 deletions

diff --git a/.babelrc b/.babelrc
index 913eb6a..911ec28 100644
--- a/.babelrc
+++ b/.babelrc
@@ -1,5 +1,8 @@
 {
-  "plugins": ["transform-es2015-modules-commonjs"],
+  "plugins": [
+    "transform-es2015-modules-commonjs",
+    "syntax-async-generators"
+  ],
   "env": {
     "test": {
       "plugins": ["istanbul"]
diff --git a/lib/internal/asyncEachOfLimit.js b/lib/internal/asyncEachOfLimit.js
new file mode 100644
index 0000000..a34bf0a
--- /dev/null
+++ b/lib/internal/asyncEachOfLimit.js
@@ -0,0 +1,59 @@
+const breakLoop = require('./breakLoop')
+
+// for async generators
+export default function asyncEachOfLimit(generator, limit, iteratee, callback) {
+    let done = false
+    let canceled = false
+    let awaiting = false
+    let running = 0
+    let idx = 0
+
+    function replenish() {
+        //console.log('replenish')
+        if (running >= limit || awaiting) return
+        //console.log('replenish awaiting')
+        awaiting = true
+        generator.next().then(({value, done: iterDone}) => {
+            //console.log('got value', value)
+            awaiting = false
+            if (iterDone) {
+                done = true;
+                if (running <= 0) {
+                    callback(null)
+                }
+                return;
+            }
+            running++
+            iteratee(value, idx, iterateeCallback)
+            idx++
+            replenish()
+        }).catch(handleError)
+    }
+
+    function iterateeCallback(err, result) {
+        //console.log('iterateeCallback')
+        if (canceled) return
+        running -= 1;
+        if (err) return handleError(err)
+
+        if (err === false) {
+            done = true;
+            canceled = true;
+        }
+
+        if (result === breakLoop || (done && running <= 0)) {
+            done = true;
+            //console.log('done')
+            return callback(null);
+        }
+        replenish()
+    }
+
+    function handleError(err) {
+        awaiting = false
+        done = true
+        callback(err)
+    }
+
+    replenish()
+}
diff --git a/lib/internal/eachOfLimit.js b/lib/internal/eachOfLimit.js
index 1d710dc..9f3f65d 100644
--- a/lib/internal/eachOfLimit.js
+++ b/lib/internal/eachOfLimit.js
@@ -3,6 +3,8 @@ import once from './once';
 
 import iterator from './iterator';
 import onlyOnce from './onlyOnce';
+import {isAsyncGenerator} from './wrapAsync'
+import asyncEachOfLimit from './asyncEachOfLimit'
 
 import breakLoop from './breakLoop';
 
@@ -15,6 +17,9 @@ export default (limit) => {
         if (!obj) {
             return callback(null);
         }
+        if (isAsyncGenerator(obj)) {
+            return asyncEachOfLimit(obj, limit, iteratee, callback)
+        }
         var nextElem = iterator(obj);
         var done = false;
         var canceled = false;
diff --git a/lib/internal/wrapAsync.js b/lib/internal/wrapAsync.js
index 2b1490c..b1be420 100644
--- a/lib/internal/wrapAsync.js
+++ b/lib/internal/wrapAsync.js
@@ -4,10 +4,14 @@ function isAsync(fn) {
     return fn[Symbol.toStringTag] === 'AsyncFunction';
 }
 
+function isAsyncGenerator(fn) {
+    return fn[Symbol.toStringTag] === 'AsyncGenerator';
+}
+
 function wrapAsync(asyncFn) {
     return isAsync(asyncFn) ? asyncify(asyncFn) : asyncFn;
 }
 
 export default wrapAsync;
 
-export { isAsync };
+export { isAsync, isAsyncGenerator };
diff --git a/package-lock.json b/package-lock.json
index 28904fa..49c303b 100644
--- a/package-lock.json
+++ b/package-lock.json
@@ -190,6 +190,7 @@
     },
     "ansi-styles": {
       "version": "2.2.1",
+      "resolved": "https://registry.npmjs.org/ansi-styles/-/ansi-styles-2.2.1.tgz",
       "integrity": "sha1-tDLdM1i2NM914eRmQ2gkBTPB3b4=",
       "dev": true
     },
@@ -1227,6 +1228,12 @@
       "integrity": "sha1-ytnK0RkbWtY0vzCuCHI5HgZHvpU=",
       "dev": true
     },
+    "babel-plugin-syntax-async-generators": {
+      "version": "6.13.0",
+      "resolved": "https://registry.npmjs.org/babel-plugin-syntax-async-generators/-/babel-plugin-syntax-async-generators-6.13.0.tgz",
+      "integrity": "sha1-a8lj67FuzLrmuStZbrfzXDQqi5o=",
+      "dev": true
+    },
     "babel-plugin-syntax-trailing-function-commas": {
       "version": "6.22.0",
       "resolved": "https://registry.npmjs.org/babel-plugin-syntax-trailing-function-commas/-/babel-plugin-syntax-trailing-function-commas-6.22.0.tgz",
@@ -3875,7 +3882,8 @@
         "ansi-regex": {
           "version": "2.1.1",
           "bundled": true,
-          "dev": true
+          "dev": true,
+          "optional": true
         },
         "aproba": {
           "version": "1.2.0",
@@ -4290,7 +4298,8 @@
         "safe-buffer": {
           "version": "5.1.1",
           "bundled": true,
-          "dev": true
+          "dev": true,
+          "optional": true
         },
         "safer-buffer": {
           "version": "2.1.2",
@@ -4346,6 +4355,7 @@
           "version": "3.0.1",
           "bundled": true,
           "dev": true,
+          "optional": true,
           "requires": {
             "ansi-regex": "^2.0.0"
           }
@@ -4389,12 +4399,14 @@
         "wrappy": {
           "version": "1.0.2",
           "bundled": true,
-          "dev": true
+          "dev": true,
+          "optional": true
         },
         "yallist": {
           "version": "3.0.2",
           "bundled": true,
-          "dev": true
+          "dev": true,
+          "optional": true
         }
       }
     },
diff --git a/package.json b/package.json
index 8f75693..213a108 100644
--- a/package.json
+++ b/package.json
@@ -23,6 +23,7 @@
     "babel-core": "^6.26.3",
     "babel-plugin-add-module-exports": "^0.2.1",
     "babel-plugin-istanbul": "^2.0.1",
+    "babel-plugin-syntax-async-generators": "^6.13.0",
     "babel-plugin-transform-es2015-modules-commonjs": "^6.26.2",
     "babel-preset-es2015": "^6.3.13",
     "babel-preset-es2017": "^6.22.0",
diff --git a/test/asyncFunctions.js b/test/asyncFunctions.js
index 894b32b..d2a1355 100644
--- a/test/asyncFunctions.js
+++ b/test/asyncFunctions.js
@@ -11,6 +11,17 @@ function supportsAsync() {
     return supported;
 }
 
+function supportsAsyncGenerators() {
+    var supported;
+    try {
+        /* eslint no-eval: 0 */
+        supported = eval('(async function * () { yield await 1 })');
+    } catch (e) {
+        supported = false;
+    }
+    return supported;
+}
+
 describe('async function support', function () {
     this.timeout(100);
 
@@ -19,4 +30,10 @@ describe('async function support', function () {
     } else {
         it('should not test async functions in this environment');
     }
+
+    if (supportsAsyncGenerators()) {
+        require('./es2017/asyncGenerators.js')();
+    } else {
+        it('should not test async generators in this environment');
+    }
 });
diff --git a/test/es2017/asyncGenerators.js b/test/es2017/asyncGenerators.js
new file mode 100644
index 0000000..4fab126
--- /dev/null
+++ b/test/es2017/asyncGenerators.js
@@ -0,0 +1,61 @@
+var async = require('../../lib');
+const {expect} = require('chai');
+const assert = require('assert');
+
+const delay = ms => new Promise(resolve => setTimeout(resolve, ms))
+
+module.exports = function () {
+    async function asyncIdentity(val) {
+        var res = await Promise.resolve(val);
+        return res;
+    }
+
+    async function * range (num) {
+        for(let i = 0; i < num; i++) {
+            await delay(1)
+            yield i
+        }
+    }
+
+    it('should handle async generators in each', (done) => {
+        const calls = []
+        async.each(range(5),
+            async (val) => {
+                calls.push(val)
+                await delay(5)
+            }, (err) => {
+                if (err) throw err
+                expect(calls).to.eql([0, 1, 2, 3, 4])
+                done()
+            }
+        )
+    });
+
+    it('should handle async generators in eachLimit', (done) => {
+        const calls = []
+        async.eachLimit(range(5), 2,
+            async (val) => {
+                calls.push(val)
+                await delay(5)
+            }, (err) => {
+                if (err) throw err
+                expect(calls).to.eql([0, 1, 2, 3, 4])
+                done()
+            }
+        )
+    });
+
+    it('should handle async generators in eachSeries', (done) => {
+        const calls = []
+        async.eachSeries(range(5),
+            async (val) => {
+                calls.push(val)
+                await delay(5)
+            }, (err) => {
+                if (err) throw err
+                expect(calls).to.eql([0, 1, 2, 3, 4])
+                done()
+            }
+        )
+    });
+}