Feature - add priority race

lib/index.js

@@ -112,6 +112,7 @@ import nextTick from './nextTick';
 import parallel from './parallel';
 import parallelLimit from './parallelLimit';
 import priorityQueue from './priorityQueue';
+import priorityRace from './priorityRace';
 import queue from './queue';
 import race from './race';
 import reduce from './reduce';
@@ -191,6 +192,7 @@ export default {
     parallel: parallel,
     parallelLimit: parallelLimit,
     priorityQueue: priorityQueue,
+    priorityRace: priorityRace,
     queue: queue,
     race: race,
     reduce: reduce,
@@ -288,6 +290,7 @@ export {
     parallel as parallel,
     parallelLimit as parallelLimit,
     priorityQueue as priorityQueue,
+    priorityRace as priorityRace,
     queue as queue,
     race as race,
     reduce as reduce,

lib/internal/priorityRace.js

@@ -0,0 +1,25 @@
+import noop from 'lodash/noop';
+import isArrayLike from 'lodash/isArrayLike';
+import slice from './slice';
+import wrapAsync from './wrapAsync';
+
+export default function _priorityRace(eachfn, tasks, callback) {
+    callback = callback || noop;
+    var results = isArrayLike(tasks) ? [] : {};
+    var outerCallback = callback;
+    var numPrioritized = tasks.filter(t => t.isPrioritized).length;
+    var numDone = 0;
+
+    eachfn(tasks, function (task, key, callback) {
+        wrapAsync(task)(function (err, result) {
+            if (arguments.length > 2) {
+                result = slice(arguments, 1);
+            }
+            results[key] = result;
+            if(task.isPrioritized && ++numDone === numPrioritized) {
+                return outerCallback(err, results)
+            }
+            callback(err);
+        });
+    }, callback);
+}

lib/priorityRace.js

@@ -0,0 +1,57 @@
+import isArray from 'lodash/isArray';
+import noop from 'lodash/noop';
+import once from './internal/once';
+import eachOf from './eachOf';
+import _priorityRace from './internal/priorityRace';
+
+/**
+ * @name priorityRace
+ * @static
+ * @memberOf module:ControlFlow
+ * @method
+ * @category Control Flow
+ * @param {Array|Iterable|Object} tasks - A collection of
+ * [async functions]{@link AsyncFunction} to run.
+ * Each async function can complete with any number of optional `result` values.
+ * Each task is non-prioritized by default but can be prioritized by setting
+ * `isPrioritized = true`.
+ * @param {Function} [callback] - An optional callback to run once all the prioritized
+ * functions have completed successfully. This function gets a results array
+ * (or object) containing all the result arguments passed to the task callbacks.
+ * Invoked with (err, results).
+ * @returns undefined
+ *
+ * @example
+ * var tasks = [
+ *     function(callback) {
+ *         setTimeout(function() {
+ *             callback(null, 'one');
+ *         }, 100);
+ *     },
+ *     function(callback) {
+ *         setTimeout(function() {
+ *             callback(null, 'two');
+ *         }, 200);
+ *     },
+ *     function(callback) {
+ *         setTimeout(function() {
+ *             callback(null, 'three');
+ *         }, 300);
+ *     }
+ * ]
+ *
+ * tasks[1].isPrioritized = true;
+ *
+ * async.priorityRace(tasks,
+ * function(err, results) {
+ *     // the results array will equal ['one','two'] because the race ended as soon
+ *     // as all the prioritzed tasks finished.
+ * });
+ *
+ */
+export default function priorityRace(tasks, callback) {
+    callback = once(callback || noop);
+    if (!isArray(tasks)) return callback(new TypeError('First argument to race must be an array of functions'));
+    if (!tasks.length) return callback();
+    _priorityRace(eachOf, tasks, callback);
+}

mocha_test/priorityRace.js

@@ -0,0 +1,92 @@
+var async = require('../lib');
+var expect = require('chai').expect;
+var assert = require('assert');
+
+describe('priorityRace', function () {
+    it('should call each function in parallel and finish when one prioritized task is done', function priorityRaceTest10(done) {
+        var finished = 0;
+        var tasks = [];
+        function eachTest(i) {
+            var index = i;
+            return function (next) {
+                finished++;
+                next(null, index);
+            };
+        }
+        for (var i = 0; i < 10; i++) {
+            tasks[i] = eachTest(i);
+        }
+        tasks[4].isPrioritized = true
+        async.priorityRace(tasks, function (err, result) {
+            assert.ifError(err);
+            //race ended when 5 finished first
+            expect(result).to.eql([0, 1, 2, 3, 4]);
+            assert.strictEqual(finished, 5);
+            done();
+        });
+    });
+    it('should call each function in parallel and finish when all prioritized tasks are done', function priorityRaceTest20(done) {
+        var finished = 0;
+        var tasks = [];
+        function eachTest(i) {
+            var index = i;
+            return function (next) {
+                finished++;
+                next(null, index);
+            };
+        }
+        for (var i = 0; i < 10; i++) {
+            tasks[i] = eachTest(i);
+        }
+        tasks[0].isPrioritized = true;
+        tasks[7].isPrioritized = true;
+        async.priorityRace(tasks, function (err, result) {
+            assert.ifError(err);
+            //race ended when both 0 and 7 finished
+            expect(result).to.eql([0, 1, 2, 3, 4, 5, 6, 7]);
+            assert.strictEqual(finished, 8);
+            done();
+        });
+    });
+    it('should callback with the first error', function priorityRaceTest30(done) {
+        var tasks = [];
+        function eachTest(i) {
+            var index = i;
+            return function (next) {
+                setTimeout(function () {
+                    next(new Error('ERR' + index));
+                }, 50 - index * 2);
+            };
+        }
+        for (var i = 0; i <= 5; i++) {
+            tasks[i] = eachTest(i);
+            tasks[0].isPrioritized = true;
+        }
+        async.priorityRace(tasks, function (err, result) {
+            assert.ok(err);
+            assert.ok(err instanceof Error);
+            assert.strictEqual(typeof result, 'undefined');
+            assert.strictEqual(err.message, 'ERR5');
+            done();
+        });
+    });
+    it('should callback when task is empty', function priorityRaceTest40(done) {
+        async.priorityRace([], function (err, result) {
+            assert.ifError(err);
+            assert.strictEqual(typeof result, 'undefined');
+            done();
+        });
+    });
+    it('should callback in error when the task arg is not an Array', function priorityRaceTest50() {
+        var errors = [];
+        async.priorityRace(null, function (err) {
+            errors.push(err);
+        });
+        async.priorityRace({}, function (err) {
+            errors.push(err);
+        });
+        assert.strictEqual(errors.length, 2);
+        assert.ok(errors[0] instanceof TypeError);
+        assert.ok(errors[1] instanceof TypeError);
+    });
+});

mocha_test/race.js

@@ -54,7 +54,7 @@ describe('race', function () {
             done();
         });
     });
-    it('should callback in error the task arg is not an Array', function raceTest40() {
+    it('should callback in error when the task arg is not an Array', function raceTest40() {
         var errors = [];
         async.race(null, function (err) {
             errors.push(err);
@@ -67,4 +67,3 @@ describe('race', function () {
         assert.ok(errors[1] instanceof TypeError);
     });
 });
-