How can I Interleave / merge async iterables?

by

There is no way to write this with a loop statement. async/await code always executes sequentially, to do things concurrently you need to use promise combinators directly. For plain promises, there’s Promise.all, for async iterators there is nothing (yet) so we need to write it on our own:

async function* combine(iterable) {
    const asyncIterators = Array.from(iterable, o => o[Symbol.asyncIterator]());
    const results = [];
    let count = asyncIterators.length;
    const never = new Promise(() => {});
    function getNext(asyncIterator, index) {
        return asyncIterator.next().then(result => ({
            index,
            result,
        }));
    }
    const nextPromises = asyncIterators.map(getNext);
    try {
        while (count) {
            const {index, result} = await Promise.race(nextPromises);
            if (result.done) {
                nextPromises[index] = never;
                results[index] = result.value;
                count--;
            } else {
                nextPromises[index] = getNext(asyncIterators[index], index);
                yield result.value;
            }
        }
    } finally {
        for (const [index, iterator] of asyncIterators.entries())
            if (nextPromises[index] != never && iterator.return != null)
                iterator.return();
        // no await here - see https://github.com/tc39/proposal-async-iteration/issues/126
    }
    return results;
}

Notice that combine does not support passing values into next or cancellation through .throw or .return.

You can call it like

(async () => {
  for await (const x of combine([a, b, c])) {
    console.log(x);
  }
})().catch(console.error);

More Related Contents:

  1. Async function returning promise, instead of value
  2. Use async await with Array.map
  3. Correct Try…Catch Syntax Using Async/Await
  4. Waiting for more than one concurrent await operation
  5. What’s wrong with awaiting a promise chain?
  6. Can I fire and forget a promise in nodejs (ES7)?
  7. async/await always returns promise
  8. How to turn this callback into a promise using async/await?
  9. Using async/await with a forEach loop
  10. What is the difference between JavaScript promises and async await?
  11. Syntax for an async arrow function
  12. How to wait for a JavaScript Promise to resolve before resuming function?
  13. How to async await in react render function?
  14. Can’t throw error from within an async promise executor function
  15. for await of VS Promise.all
  16. JavaScript array .reduce with async/await
  17. Why do I need to await an async function when it is not supposedly returning a Promise?
  18. Slowdown due to non-parallel awaiting of promises in async generators
  19. using async await and .then together
  20. How to implement a pseudo blocking async queue in JS/TS?
  21. Do Javascript promises block the stack
  22. What are asynchronous functions in JavaScript? What is “async” and “await” in JavaScript?
  23. Wait promise inside for loop
  24. Difference of using async / await vs promises?
  25. Async Await map not awaiting async function to complete inside map function before mapping next item
  26. Implement async/await in sort function of arrays javascript
  27. Can I catch an error from async without using await?
  28. Node JS Promise.all and forEach
  29. How to do promise.all for array of array of promises?
  30. Are JavaScript forever-pending promises bad?

Leave a Comment