GatherUnordered.js

gatherUnordered.js

/*
 * gather unordered takes an Array of promise's running them in parallel returning a single promise of their result and cancel function.
 *
 * If a error is encountered the first error will be returned, remaining unordered thunks will not be executed.
 *
 * @inputThunks Array[() => Promise[A]] an array of promises not yet started, thunks () => Promise
 * @parallism number of promises to run at once
 * @returns Object {
 *   promise: Promise[Array[A]]
 *   cancel: Function() to cancel
 *  }
 */
export default function gatherUnordered(inputThunks, parallelism) {
  let running = true;
  const size = inputThunks.length;
  const allCompleted = [];

  let _resolve, _reject;
  const promise = new Promise((resolve, reject) => {
    _resolve = resolve;
    _reject = reject;
  });

  function go() {
    if (!running) {
      _reject('cancelled');
    } else if (allCompleted.length === size) {
      _resolve(allCompleted);
    } else {
      const thunk = inputThunks.shift();
      return thunk && thunk().then(result => {
        allCompleted.push(result);
        return go();
      }).catch(e => _reject(e));
    }
  }

  for (let i = 0; i < parallelism; i++) {
    go();
  }

  const self = {
    promise: promise,
    cancel: () => {
      running = false;
    },
    catch: (fn) => {
      promise.catch(fn);
      return self;
    }
  };
  return self;
}