JS Async Generator Race

sortPromisesByResolutionTime.js

const genToAsyncGen = async iteration => ({
  value: await (await iteration).value,
  done: (await iteration).done,
});

const iterableToIterator = (iterable = []) => {
  if (iterable[Symbol.asyncIterator])
    return iterable[Symbol.asyncIterator]();
  if (iterable[Symbol.iterator])
    return iterable[Symbol.iterator]();
  if (iterable.next) return iterable;
  return [].values();
}

// works with iterable and asyncIterable

exports.asRace = (concurrencyLimit = 10) => (
  async function* innerRace(iterable) {
    const iterator = iterableToIterator(iterable);
    let numWorkers = concurrencyLimit;
    const workInProgress = new Set();
    const finished = new Set();
    do {
      while (workInProgress.size !== numWorkers) {
        const task = genToAsyncGen(
          iterator.next()
        );
        workInProgress.add(task);
        task.then(({ done, value }) => {
          workInProgress.delete(task);
          if (done) numWorkers -= 1;
          else finished.add(value);
        }).catch(e => {
          workInProgress.delete(task);
          finished.add(e)
        });
      }
      if (workInProgress.size)
        await Promise.race(workInProgress);
      yield * finished;
      finished.clear();
    } while (numWorkers);
  }
);


const toPromise = async value => value;


// works only with iterable
async function* sortByResolution(iterable) {
  const inProgress = new Set(iterable);
  const finished = new Set();
  do {
    Array.from(inProgress).map(task => toPromise(task)
      .then(v => finished.add(v))
      .catch(e => finished.add(e))
      .finally(() => inProgress.delete(task)));

    if (inProgress.size) {
      await Promise.race(inProgress);
    }
    yield * finished;
    finished.clear();
  } while (inProgress.size);
}

exports.sortByResolution = sortByResolution;