Process items from Iterable in parallel with given concurrency factor

index.js

async function eachParallel(iterable, action, concurrencyFactor) {
  const evt = new EventEmitter();
  let runningCount = 0;
  let cancelled = false;
  const errors = new Map();
  let ended = false;

  const emit = evt.emit.bind(evt);
  const on = evt.on.bind(evt);
  const until = eventName => new Promise(res => evt.once(eventName, res));

  async function run(item) {
    try {
      runningCount += 1;
      await action(item);
    } catch (err) {
      errors.set(item, err);
      cancelled = true;
    } finally {
      runningCount -= 1;
      emit('itemProcessed');
    }
  }

  async function startConsuming() {
    for await (const item of iterable) {
      while (runningCount >= concurrencyFactor) {
        await until('itemProcessed');
      }

      if (cancelled) break;
      run(item);
    }
    ended = true;
    emit('end');
  }

  startConsuming();

  return new Promise((res, rej) => {
    function maybeFinalize() {
      if (ended && runningCount === 0) {
        if (errors.size > 0) rej(errors);
        else res();
      }
    }

    on('itemProcessed', maybeFinalize);
    on('end', maybeFinalize);
  });
}