Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: await-able Async methods #1572

Merged
merged 25 commits into from Oct 1, 2018
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
8 changes: 6 additions & 2 deletions lib/each.js
@@ -1,6 +1,7 @@
import eachOf from './eachOf';
import withoutIndex from './internal/withoutIndex';
import wrapAsync from './internal/wrapAsync'
import awaitify from './internal/awaitify'

/**
* Applies the function `iteratee` to each item in `coll`, in parallel.
Expand All @@ -25,6 +26,7 @@ import wrapAsync from './internal/wrapAsync'
* If you need the index, use `eachOf`.
* @param {Function} [callback] - A callback which is called when all
* `iteratee` functions have finished, or an error occurs. Invoked with (err).
* @returns {Promise} a promise, if a callback is omitted
* @example
*
* // assuming openFiles is an array of file names and saveFile is a function
Expand Down Expand Up @@ -59,6 +61,8 @@ import wrapAsync from './internal/wrapAsync'
* }
* });
*/
export default function eachLimit(coll, iteratee, callback) {
eachOf(coll, withoutIndex(wrapAsync(iteratee)), callback);
function eachLimit(coll, iteratee, callback) {
return eachOf(coll, withoutIndex(wrapAsync(iteratee)), callback);
}

export default awaitify(eachLimit, 3)
7 changes: 5 additions & 2 deletions lib/eachLimit.js
@@ -1,6 +1,7 @@
import eachOfLimit from './internal/eachOfLimit';
import withoutIndex from './internal/withoutIndex';
import wrapAsync from './internal/wrapAsync';
import awaitify from './internal/awaitify'

/**
* The same as [`each`]{@link module:Collections.each} but runs a maximum of `limit` async operations at a time.
Expand All @@ -21,7 +22,9 @@ import wrapAsync from './internal/wrapAsync';
* Invoked with (item, callback).
* @param {Function} [callback] - A callback which is called when all
* `iteratee` functions have finished, or an error occurs. Invoked with (err).
* @returns {Promise} a promise, if a callback is omitted
*/
export default function eachLimit(coll, limit, iteratee, callback) {
eachOfLimit(limit)(coll, withoutIndex(wrapAsync(iteratee)), callback);
function eachLimit(coll, limit, iteratee, callback) {
return eachOfLimit(limit)(coll, withoutIndex(wrapAsync(iteratee)), callback);
}
export default awaitify(eachLimit, 4)
6 changes: 4 additions & 2 deletions lib/eachOf.js
Expand Up @@ -6,6 +6,7 @@ import noop from './internal/noop';
import once from './internal/once';
import onlyOnce from './internal/onlyOnce';
import wrapAsync from './internal/wrapAsync';
import awaitify from './internal/awaitify'

// eachOf implementation optimized for array-likes
function eachOfArrayLike(coll, iteratee, callback) {
Expand Down Expand Up @@ -56,6 +57,7 @@ var eachOfGeneric = doLimit(eachOfLimit, Infinity);
* Invoked with (item, key, callback).
* @param {Function} [callback] - A callback which is called when all
* `iteratee` functions have finished, or an error occurs. Invoked with (err).
* @returns {Promise} a promise, if a callback is omitted
* @example
*
* var obj = {dev: "/dev.json", test: "/test.json", prod: "/prod.json"};
Expand All @@ -77,7 +79,7 @@ var eachOfGeneric = doLimit(eachOfLimit, Infinity);
* doSomethingWith(configs);
* });
*/
export default function(coll, iteratee, callback) {
export default awaitify((coll, iteratee, callback) => {
var eachOfImplementation = isArrayLike(coll) ? eachOfArrayLike : eachOfGeneric;
eachOfImplementation(coll, wrapAsync(iteratee), callback);
}
}, 3)
8 changes: 6 additions & 2 deletions lib/eachOfLimit.js
@@ -1,5 +1,6 @@
import _eachOfLimit from './internal/eachOfLimit';
import wrapAsync from './internal/wrapAsync';
import awaitify from './internal/awaitify'

/**
* The same as [`eachOf`]{@link module:Collections.eachOf} but runs a maximum of `limit` async operations at a
Expand All @@ -20,7 +21,10 @@ import wrapAsync from './internal/wrapAsync';
* Invoked with (item, key, callback).
* @param {Function} [callback] - A callback which is called when all
* `iteratee` functions have finished, or an error occurs. Invoked with (err).
* @returns {Promise} a promise, if a callback is omitted
*/
export default function eachOfLimit(coll, limit, iteratee, callback) {
_eachOfLimit(limit)(coll, wrapAsync(iteratee), callback);
function eachOfLimit(coll, limit, iteratee, callback) {
return _eachOfLimit(limit)(coll, wrapAsync(iteratee), callback);
}

export default awaitify(eachOfLimit, 4)
4 changes: 3 additions & 1 deletion lib/eachOfSeries.js
@@ -1,5 +1,6 @@
import eachOfLimit from './eachOfLimit';
import doLimit from './internal/doLimit';
import awaitify from './internal/awaitify'

/**
* The same as [`eachOf`]{@link module:Collections.eachOf} but runs only a single async operation at a time.
Expand All @@ -17,5 +18,6 @@ import doLimit from './internal/doLimit';
* Invoked with (item, key, callback).
* @param {Function} [callback] - A callback which is called when all `iteratee`
* functions have finished, or an error occurs. Invoked with (err).
* @returns {Promise} a promise, if a callback is omitted
*/
export default doLimit(eachOfLimit, 1);
export default awaitify(doLimit(eachOfLimit, 1), 3);
4 changes: 3 additions & 1 deletion lib/eachSeries.js
@@ -1,5 +1,6 @@
import eachLimit from './eachLimit';
import doLimit from './internal/doLimit';
import awaitify from './internal/awaitify'

/**
* The same as [`each`]{@link module:Collections.each} but runs only a single async operation at a time.
Expand All @@ -19,5 +20,6 @@ import doLimit from './internal/doLimit';
* Invoked with (item, callback).
* @param {Function} [callback] - A callback which is called when all
* `iteratee` functions have finished, or an error occurs. Invoked with (err).
* @returns {Promise} a promise, if a callback is omitted
*/
export default doLimit(eachLimit, 1);
export default awaitify(doLimit(eachLimit, 1), 3);
25 changes: 25 additions & 0 deletions lib/internal/awaitify.js
@@ -0,0 +1,25 @@
export const ASYNC_FN = Symbol('asyncFunction')

// conditionally promisify a function.
// only return a promise if a callback is omitted
export default function awaitify (asyncFn, arity) {
function awaitable(...args) {
if (args.length === arity || typeof args[arity - 1] === 'function') {
return asyncFn.apply(this, args)
}

return new Promise((resolve, reject) => {
args.push((err, ...cbArgs) => {
if (err) return reject(err)
resolve(cbArgs.length > 1 ? cbArgs : cbArgs[0])
})
asyncFn.apply(this, args)
})
}

awaitable[Symbol.toStringTag] = 'AsyncFunction'
Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is a bit of a hack, not sure if we should do this.

awaitable[ASYNC_FN] = true
Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I intend to use this so we can avoid an additional layer of wrapping in wrapAsync.

awaitable.displayName = `awaitable(${asyncFn.name})`

return awaitable
}
19 changes: 19 additions & 0 deletions lib/internal/promiseCallback.js
@@ -0,0 +1,19 @@
const PROMISE_SYMBOL = Symbol('promiseCallback')

function promiseCallback () {
Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is the older strategy. It's a bit more clunky, but doesn't add another frame to the call stack for each method.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I had to use this in a couple places where the awaitify wrapper couldn't work (usually due to optional arguments).

let resolve, reject
function callback (err, ...args) {
if (err) return reject(err)
resolve(args.length > 1 ? args : args[0])
}

callback[PROMISE_SYMBOL] = new Promise((res, rej) => {
resolve = res,
reject = rej
})

return callback
}


export { promiseCallback, PROMISE_SYMBOL }
3 changes: 3 additions & 0 deletions test/asyncFunctions.js
Expand Up @@ -27,6 +27,9 @@ describe('async function support', function () {

if (supportsAsync()) {
require('./es2017/asyncFunctions.js').call(this);
describe('awaitable functions', () => {
require('./es2017/awaitableFunctions.js').call(this);
});
} else {
it('should not test async functions in this environment');
}
Expand Down
59 changes: 59 additions & 0 deletions test/es2017/awaitableFunctions.js
@@ -0,0 +1,59 @@
var async = require('../../lib');
const {expect} = require('chai');
const assert = require('assert');


module.exports = function () {
async function asyncIdentity(val) {
var res = await Promise.resolve(val);
return res;
}

const input = [1, 2, 3];
const inputObj = {a: 1, b: 2, c: 3};

it('should asyncify async functions', (done) => {
async.asyncify(asyncIdentity)(42, (err, val) => {
assert(val === 42);
done(err);
})
});


/*
* Collections
*/

it('should return a Promise: each', async () => {
const calls = []
await async.each(input, async val => { calls.push(val) });
expect(calls).to.eql([1, 2, 3])
expect(async.each(input, asyncIdentity) instanceof Promise).to.equal(true)
});
it('should return a Promise: eachSeries', async () => {
const calls = []
await async.eachSeries(input, async val => { calls.push(val) });
expect(calls).to.eql([1, 2, 3])
});
it('should return a Promise: eachLimit', async () => {
const calls = []
await async.eachLimit(input, 1, async val => { calls.push(val) });
expect(calls).to.eql([1, 2, 3])
});

it('should return a Promise: eachOf', async () => {
const calls = []
await async.eachOf(inputObj, async (...args) => { calls.push(args) });
expect(calls).to.eql([[1, 'a'], [2, 'b'], [3, 'c']])
});
it('should return a Promise: eachOfSeries', async () => {
const calls = []
await async.eachOfSeries(inputObj, async (...args) => { calls.push(args) });
expect(calls).to.eql([[1, 'a'], [2, 'b'], [3, 'c']])
});
it('should return a Promise: eachOfLimit', async () => {
const calls = []
await async.eachOfLimit(inputObj, 1, async (...args) => { calls.push(args) });
expect(calls).to.eql([[1, 'a'], [2, 'b'], [3, 'c']])
});
};