Please support this book: buy it or donate

5. Asynchronous iteration



This chapter explains the ECMAScript proposal “Asynchronous Iteration” by Domenic Denicola and Kevin Smith.

5.1. Asynchronous iteration

With ECMAScript 6, JavaScript got built-in support for synchronously iterating over data. But what about data that is delivered asynchronously? For example, lines of text, read asynchronously from a file or an HTTP connection.

This proposal brings support for that kind of data. Before we go into it, let’s first recap synchronous iteration.

5.1.1. Synchronous iteration

Synchronous iteration was introduced with ES6 and works as follows:

  • Iterable: an object that signals that it can be iterated over, via a method whose key is Symbol.iterator.
  • Iterator: an object returned by invoking Symbol.iterator on an iterable. It wraps each iterated element in an object and returns it via its method next() – one at a time.
  • IteratorResult: an object returned by next(). Property value contains an iterated element, property done is true after the last element (value can usually be ignored then; it’s almost always undefined). I’ll demonstrate via an Array:
  1. > const iterable = ['a', 'b'];
  2. > const iterator = iterable[Symbol.iterator]();
  3. > iterator.next()
  4. { value: 'a', done: false }
  5. > iterator.next()
  6. { value: 'b', done: false }
  7. > iterator.next()
  8. { value: undefined, done: true }

5.1.2. Asynchronous iteration

The problem is that the previously explained way of iterating is synchronous, it doesn’t work for asynchronous sources of data. For example, in the following code, readLinesFromFile() cannot deliver its asynchronous data via synchronous iteration:

  1. for (const line of readLinesFromFile(fileName)) {
  2. console.log(line);
  3. }

The proposal specifies a new protocol for iteration that works asynchronously:

  • Async iterables are marked via Symbol.asyncIterator.
  • Method next() of an async iterator returns Promises for IteratorResults (vs. IteratorResults directly). You may wonder whether it would be possible to instead use a synchronous iterator that returns one Promise for each iterated element. But that is not enough – whether or not iteration is done is generally determined asynchronously.

Using an asynchronous iterable looks as follows. Function createAsyncIterable() is explained later. It converts its synchronously iterable parameter into an async iterable.

  1. const asyncIterable = createAsyncIterable(['a', 'b']);
  2. const asyncIterator = asyncIterable[Symbol.asyncIterator]();
  3. asyncIterator.next()
  4. .then(iterResult1 => {
  5. console.log(iterResult1); // { value: 'a', done: false }
  6. return asyncIterator.next();
  7. })
  8. .then(iterResult2 => {
  9. console.log(iterResult2); // { value: 'b', done: false }
  10. return asyncIterator.next();
  11. })
  12. .then(iterResult3 => {
  13. console.log(iterResult3); // { value: undefined, done: true }
  14. });

Within an asynchronous function, you can process the results of the Promises via await and the code becomes simpler:

  1. async function f() {
  2. const asyncIterable = createAsyncIterable(['a', 'b']);
  3. const asyncIterator = asyncIterable[Symbol.asyncIterator]();
  4. console.log(await asyncIterator.next());
  5. // { value: 'a', done: false }
  6. console.log(await asyncIterator.next());
  7. // { value: 'b', done: false }
  8. console.log(await asyncIterator.next());
  9. // { value: undefined, done: true }
  10. }

5.1.3. The interfaces for async iteration

In TypeScript notation, the interfaces look as follows.

  1. interface AsyncIterable {
  2. [Symbol.asyncIterator]() : AsyncIterator;
  3. }
  4. interface AsyncIterator {
  5. next() : Promise<IteratorResult>;
  6. }
  7. interface IteratorResult {
  8. value: any;
  9. done: boolean;
  10. }

5.2. for-await-of

The proposal also specifies an asynchronous version of the for-of loop: for-await-of:

  1. async function f() {
  2. for await (const x of createAsyncIterable(['a', 'b'])) {
  3. console.log(x);
  4. }
  5. }
  6. // Output:
  7. // a
  8. // b

5.2.1. for-await-of and rejections

Similarly to how await works in async functions, the loop throws an exception if next() returns a rejection:

  1. function createRejectingIterable() {
  2. return {
  3. [Symbol.asyncIterator]() {
  4. return this;
  5. },
  6. next() {
  7. return Promise.reject(new Error('Problem!'));
  8. },
  9. };
  10. }
  11. (async function () { // (A)
  12. try {
  13. for await (const x of createRejectingIterable()) {
  14. console.log(x);
  15. }
  16. } catch (e) {
  17. console.error(e);
  18. // Error: Problem!
  19. }
  20. })(); // (B)

Note that we have just used an Immediately Invoked Async Function Expression (IIAFE, pronounced “yaffee”). It starts in line (A) and ends in line (B). We need to do that because for-of-await doesn’t work at the top level of modules and scripts. It does work everywhere where await can be used. Namely, in async functions and async generators (which are explained later).

5.2.2. for-await-of and synchronous iterables

Synchronous iterables return synchronous iterators, whose method next() returns {value, done} objects. for-await-of handles synchronous iterables by converting them to asynchronous iterables. Each iterated value is converted to a Promise (or left unchanged if it already is a Promise) via Promise.resolve(). That is, for-await-of works for iterables over Promises and over normal values. The conversion looks like this:

  1. const nextResult = Promise.resolve(valueOrPromise)
  2. .then(x => ({ value: x, done: false }));

Two more ways of looking at the conversion are:

  • Iterable<Promise<T>> becomes AsyncIterable<T>
  • The following object
  1. { value: Promise.resolve(123), done: false }

is converted to

  1. Promise.resolve({ value: 123, done: false })

Therefore, the following two statements are roughly similar.

  1. for (const x of await Promise.all(syncIterableOverPromises));
  2. for await (const x of syncIterableOverPromises);

The second statement is faster, because Promise.all() only creates the Promise for the Array after all Promises in syncIterableOverPromises are fulfilled. And for-of has to await that Promise. In contrast, for-await-of starts processing as soon as the first Promise is fulfilled.

5.2.3. Example: for-await-of with a sync iterable

Iterating over a sync iterable over Promises:

  1. async function main() {
  2. const syncIterable = [
  3. Promise.resolve('a'),
  4. Promise.resolve('b'),
  5. ];
  6. for await (const x of syncIterable) {
  7. console.log(x);
  8. }
  9. }
  10. main();
  11. // Output:
  12. // a
  13. // b

Iterating over a sync iterable over normal values:

  1. async function main() {
  2. for await (const x of ['c', 'd']) {
  3. console.log(x);
  4. }
  5. }
  6. main();
  7. // Output:
  8. // c
  9. // d

5.3. Asynchronous generators

Normal (synchronous) generators help with implementing synchronous iterables. Asynchronous generators do the same for asynchronous iterables.

For example, we have previously used the function createAsyncIterable(syncIterable) which converts a syncIterable into an asynchronous iterable. This is how you would implement this function via an async generator:

  1. async function* createAsyncIterable(syncIterable) {
  2. for (const elem of syncIterable) {
  3. yield elem;
  4. }
  5. }

Note the asterisk after function:

  • A normal function is turned into a normal generator by putting an asterisk after function.
  • An async function is turned into an async generator by doing the same. How do async generators work?

  • A normal generator returns a generator object genObj. Each invocation genObj.next() returns an object {value,done} that wraps a yielded value.

  • An async generator returns a generator object genObj. Each invocation genObj.next() returns a Promise for an object {value,done} that wraps a yielded value.

5.3.1. Queuing next() invocations

The JavaScript engine internally queues invocations of next() and feeds them to an async generator once it is ready. That is, after calling next(), you can call again, right away; you don’t have to wait for the Promise it returns to be settled. In most cases, though, you do want to wait for the settlement, because you need the value of done in order to decide whether to call next() again or not. That’s how the for-await-of loop works.

Use cases for calling next() several times without waiting for settlements include:

Use case: Retrieving Promises to be processed via Promise.all(). If you know how many elements there are in an async iterable, you don’t need to check done.

  1. const asyncGenObj = createAsyncIterable(['a', 'b']);
  2. const [{value:v1},{value:v2}] = await Promise.all([
  3. asyncGenObj.next(), asyncGenObj.next()
  4. ]);
  5. console.log(v1, v2); // a b

Use case: Async generators as sinks for data, where you don’t always need to know when they are done.

  1. const writer = openFile('someFile.txt');
  2. writer.next('hello'); // don’t wait
  3. writer.next('world'); // don’t wait
  4. await writer.return(); // wait for file to close

Acknowledgement: Thanks to [@domenic] and [@zenparsing] for these use cases.

5.3.2. await in async generators

You can use await and for-await-of inside async generators. For example:

  1. async function* prefixLines(asyncIterable) {
  2. for await (const line of asyncIterable) {
  3. yield '> ' + line;
  4. }
  5. }

One interesting aspect of combining await and yield is that await can’t stop yield from returning a Promise, but it can stop that Promise from being settled:

  1. async function* asyncGenerator() {
  2. console.log('Start');
  3. const result = await doSomethingAsync(); // (A)
  4. yield 'Result: '+result; // (B)
  5. console.log('Done');
  6. }

This is the context in which this code is executed:

  • Every asynchronous generator has a queue with Promises to be settled yield or throw.
  • When .next() is called, the following steps are taken:

    • It queues a Promise.
    • Unless the async generator is already running, it resumes it and waits for it to be finished. (It may finish via yield, throw, return, await.)
    • Then it returns the Promise. Recall that the result of a settled Promise is delivered asynchronously. Therefore, the earliest delivery is during the next tick. What does this mean for asyncGenerator()?
  • Execution starts and progresses until line A, when the generator pauses (due to await) and execution reverts back to .next(), which returns a Promise P.

  • When the Promise returned by doSomethingAsync() is fulfilled, the generator is resumed and fulfills P with result (via yield in line B). Then the generator is suspended.
  • When it is resumed via .next(), it fulfills the Promise at the front of the queue via an implicit return undefined at the end. That means that line A and B correspond (roughly) to this code:
  1. doSomethingAsync()
  2. .then(result => {
  3. const {resolve} = promiseQueue.dequeue();
  4. resolve({
  5. value: 'Result: '+result,
  6. done: false,
  7. });
  8. });

If you want to dig deeper – this is a rough approximation of how async generators work:

  1. const BUSY = Symbol('BUSY');
  2. const COMPLETED = Symbol('COMPLETED');
  3. function asyncGenerator() {
  4. const settlers = [];
  5. let step = 0;
  6. return {
  7. [Symbol.asyncIterator]() {
  8. return this;
  9. },
  10. next() {
  11. return new Promise((resolve, reject) => {
  12. settlers.push({resolve, reject});
  13. this._run();
  14. });
  15. }
  16. _run() {
  17. setTimeout(() => {
  18. if (step === BUSY || settlers.length === 0) {
  19. return;
  20. }
  21. const currentSettler = settlers.shift();
  22. try {
  23. switch (step) {
  24. case 0:
  25. step = BUSY;
  26. console.log('Start');
  27. doSomethingAsync()
  28. .then(result => {
  29. currentSettler.resolve({
  30. value: 'Result: '+result,
  31. done: false,
  32. });
  33. // We are not busy, anymore
  34. step = 1;
  35. this._run();
  36. })
  37. .catch(e => currentSettler.reject(e));
  38. break;
  39. case 1:
  40. console.log('Done');
  41. currentSettler.resolve({
  42. value: undefined,
  43. done: true,
  44. });
  45. step = COMPLETED;
  46. this._run();
  47. break;
  48. case COMPLETED:
  49. currentSettler.resolve({
  50. value: undefined,
  51. done: true,
  52. });
  53. this._run();
  54. break;
  55. }
  56. }
  57. catch (e) {
  58. currentSettler.reject(e);
  59. }
  60. }, 0);
  61. }
  62. }
  63. }

This code assumes that next() is always called without arguments. A complete implementation would have to queue arguments, too.

5.3.3. yield* in async generators

yield* in async generators works analogously to how it works in normal generators – like a recursive invocation:

  1. async function* gen1() {
  2. yield 'a';
  3. yield 'b';
  4. return 2;
  5. }
  6. async function* gen2() {
  7. const result = yield* gen1(); // (A)
  8. // result === 2
  9. }

In line (A), gen2() calls gen1(), which means that all elements yielded by gen1() are yielded by gen2():

  1. (async function () {
  2. for await (const x of gen2()) {
  3. console.log(x);
  4. }
  5. })();
  6. // Output:
  7. // a
  8. // b

The operand of yield* can be any async iterable. Sync iterables are automatically converted to async iterables, just like with for-await-of.

5.3.4. Errors

In normal generators, next() can throw exceptions. In async generators, next() can reject the Promise it returns:

  1. async function* asyncGenerator() {
  2. // The following exception is converted to a rejection
  3. throw new Error('Problem!');
  4. }
  5. asyncGenerator().next()
  6. .catch(err => console.log(err)); // Error: Problem!

Converting exceptions to rejections is similar to how async functions work.

5.3.5. Async function vs. async generator function

Async function:

  • Returns immediately with a Promise.
  • That Promise is fulfilled via return and rejected via throw.
  1. (async function () {
  2. return 'hello';
  3. })()
  4. .then(x => console.log(x)); // hello
  5. (async function () {
  6. throw new Error('Problem!');
  7. })()
  8. .catch(x => console.error(x)); // Error: Problem!

Async generator function:

  • Returns immediately with an async iterable.
  • Every invocation of next() returns a Promise. yield x fulfills the “current” Promise with {value: x, done: false}. throw err rejects the “current” Promise with err.
  1. async function* gen() {
  2. yield 'hello';
  3. }
  4. const genObj = gen();
  5. genObj.next().then(x => console.log(x));
  6. // { value: 'hello', done: false }

5.4. Examples

The source code for the examples is available via the repository async-iter-demo on GitHub.

5.4.1. Using asynchronous iteration via Babel

The example repo uses babel-node to run its code. This is how it configures Babel in its package.json:

  1. {
  2. "dependencies": {
  3. "babel-preset-env": "···",
  4. "babel-plugin-transform-async-generator-functions": "···",
  5. ···
  6. },
  7. "babel": {
  8. "presets": [
  9. [
  10. "env",
  11. {
  12. "targets": {
  13. "node": "current"
  14. }
  15. }
  16. ]
  17. ],
  18. "plugins": [
  19. "transform-async-generator-functions"
  20. ]
  21. },
  22. ···
  23. }

5.4.2. Example: turning an async iterable into an Array

Function takeAsync() collects all elements of asyncIterable in an Array. I don’t use for-await-of in this case, I invoke the async iteration protocol manually. I also don’t close asyncIterable if I’m finished before the iterable is done.

  1. /**
  2. * @returns a Promise for an Array with the elements
  3. * in `asyncIterable`
  4. */
  5. async function takeAsync(asyncIterable, count=Infinity) {
  6. const result = [];
  7. const iterator = asyncIterable[Symbol.asyncIterator]();
  8. while (result.length < count) {
  9. const {value,done} = await iterator.next();
  10. if (done) break;
  11. result.push(value);
  12. }
  13. return result;
  14. }

This is the test for takeAsync():

  1. test('Collect values yielded by an async generator', async function() {
  2. async function* gen() {
  3. yield 'a';
  4. yield 'b';
  5. yield 'c';
  6. }
  7. assert.deepStrictEqual(await takeAsync(gen()), ['a', 'b', 'c']);
  8. assert.deepStrictEqual(await takeAsync(gen(), 3), ['a', 'b', 'c']);
  9. assert.deepStrictEqual(await takeAsync(gen(), 2), ['a', 'b']);
  10. assert.deepStrictEqual(await takeAsync(gen(), 1), ['a']);
  11. assert.deepStrictEqual(await takeAsync(gen(), 0), []);
  12. });

Note how nicely async functions work together with the mocha test framework: for asynchronous tests, the second parameter of test() can return a Promise.

5.4.3. Example: a queue as an async iterable

The example repo also has an implementation for an asynchronous queue, called AsyncQueue. Its implementation is relatively complex, which is why I don’t show it here. This is the test for AsyncQueue:

  1. test('Enqueue before dequeue', async function() {
  2. const queue = new AsyncQueue();
  3. queue.enqueue('a');
  4. queue.enqueue('b');
  5. queue.close();
  6. assert.deepStrictEqual(await takeAsync(queue), ['a', 'b']);
  7. });
  8. test('Dequeue before enqueue', async function() {
  9. const queue = new AsyncQueue();
  10. const promise = Promise.all([queue.next(), queue.next()]);
  11. queue.enqueue('a');
  12. queue.enqueue('b');
  13. return promise.then(arr => {
  14. const values = arr.map(x => x.value);
  15. assert.deepStrictEqual(values, ['a', 'b']);
  16. });
  17. });

5.4.4. Example: reading text lines asynchronously

Let’s implement code that reads text lines asynchronously. We’ll do it in three steps.

Step 1: read text data in chunks via the Node.js ReadStream API (which is based on callbacks) and push it into an AsyncQueue (which was introduced in the previous section).

  1. /**
  2. * Creates an asynchronous ReadStream for the file whose name
  3. * is `fileName` and feeds it into an AsyncQueue that it returns.
  4. *
  5. * @returns an async iterable
  6. */
  7. function readFile(fileName) {
  8. const queue = new AsyncQueue();
  9. const readStream = createReadStream(fileName,
  10. { encoding: 'utf8', bufferSize: 1024 });
  11. readStream.on('data', buffer => {
  12. const str = buffer.toString('utf8');
  13. queue.enqueue(str);
  14. });
  15. readStream.on('end', () => {
  16. // Signal end of output sequence
  17. queue.close();
  18. });
  19. return queue;
  20. }

Step 2: Use for-await-of to iterate over the chunks of text and yield lines of text.

  1. /**
  2. * Turns a sequence of text chunks into a sequence of lines
  3. * (where lines are separated by newlines)
  4. *
  5. * @returns an async iterable
  6. */
  7. async function* splitLines(chunksAsync) {
  8. let previous = '';
  9. for await (const chunk of chunksAsync) {
  10. previous += chunk;
  11. let eolIndex;
  12. while ((eolIndex = previous.indexOf('\n')) >= 0) {
  13. const line = previous.slice(0, eolIndex);
  14. yield line;
  15. previous = previous.slice(eolIndex+1);
  16. }
  17. }
  18. if (previous.length > 0) {
  19. yield previous;
  20. }
  21. }

Step 3: combine the two previous functions. We first feed chunks of text into a queue via readFile() and then convert that queue into an async iterable over lines of text via splitLines().

  1. /**
  2. * @returns an async iterable
  3. */
  4. function readLines(fileName) {
  5. // `queue` is an async iterable
  6. const queue = readFile(fileName);
  7. return splitLines(queue);
  8. }

Lastly, this is how you’d use readLines() from within a Node.js script:

  1. (async function () {
  2. const fileName = process.argv[2];
  3. for await (const line of readLines(fileName)) {
  4. console.log('>', line);
  5. }
  6. })();

5.5. WHATWG Streams are async iterables

WHATWG streams are async iterables, meaning that you can use for-await-of to process them:

  1. const rs = openReadableStream();
  2. for await (const chunk of rs) {
  3. ···
  4. }

5.6. The specification of asynchronous iteration

The spec introduces several new concepts and entities:

5.6.1. Async generators

If you want to understand how async generators work, it’s best to start with Sect. “AsyncGenerator Abstract Operations”. The key to understanding async generators is understanding how queuing works.

Two internal properties of async generator objects play important roles w.r.t. queuing:

  • [[AsyncGeneratorState]] contains the state the generator is currently in: "suspendedStart", "suspendedYield", "executing", "completed" (it is undefined before it is fully initialized)
  • [[AsyncGeneratorQueue]] holds pending invocations of next/throw/return. Each queue entry contains two fields:

    • [[Completion]]: the parameter of next(), throw() or return() that lead to the entry being enqueued. The type of the completion (normal, throw, return) indicates which method call created the entry and determines what happens after dequeuing.
    • [[Capability]]: the PromiseCapability of the pending Promise. The queue is managed mainly via two operations:
  • Enqueuing happens via AsyncGeneratorEnqueue(). This is the operation that is called by next(), return() and throw(). It adds an entry to the AsyncGeneratorQueue. Then AsyncGeneratorResumeNext() is called, but only if the generator’s state isn’t "executing":

    • Therefore, if a generator calls next(), return() or throw() from inside itself then the effects of that call will be delayed.
    • await leads to a suspension of the generator, but its state remains "executing". Hence, it will not be resumed by AsyncGeneratorEnqueue().
  • Dequeuing happens via AsyncGeneratorResumeNext(). AsyncGeneratorResumeNext() is invoked after enqueuing, but also after settling a queued Promise (e.g. via yield), because there may now be new queued pending Promises, allowing execution to continue. If the queue is empty, return immediately. Otherwise, the current Promise is the first element of the queue:
    • If the async generator was suspended by yield, it is resumed and continues to run. The current Promise is later settled via AsyncGeneratorResolve() or AsyncGeneratorReject().
    • If the generator is already completed, this operation calls AsyncGeneratorResolve() and AsyncGeneratorReject() itself, meaning that all queued pending Promises will eventually be settled.

5.6.2. Async-from-Sync Iterator Objects

To get an async iterator from an object iterable, you call GetIterator(iterable, async) (async is a symbol). If iterable doesn’t have a method Symbol.asyncIterator, GetIterator() retrieves a sync iterator via method iterableSymbol.iterator and converts it to an async iterator via CreateAsyncFromSyncIterator().

5.6.3. The for-await-of loop

for-await-of works almost exactly like for-of, but there is an await whenever the contents of an IteratorResult are accessed. You can see that by looking at Sect. “Runtime Semantics: ForIn/OfBodyEvaluation”. Notably, iterators are closed similarly, via IteratorClose(), towards the end of this section.

5.7. Alternatives to async iteration

Let’s look at two alternatives to async iteration for processing async data.

5.7.1. Alternative 1: Communicating Sequential Processes (CSP)

The following code demonstrates the CSP library js-csp:

  1. var csp = require('js-csp');
  2. function* player(name, table) {
  3. while (true) {
  4. var ball = yield csp.take(table); // dequeue
  5. if (ball === csp.CLOSED) {
  6. console.log(name + ": table's gone");
  7. return;
  8. }
  9. ball.hits += 1;
  10. console.log(name + " " + ball.hits);
  11. yield csp.timeout(100); // wait
  12. yield csp.put(table, ball); // enqueue
  13. }
  14. }
  15. csp.go(function* () {
  16. var table = csp.chan(); // (A)
  17. csp.go(player, ["ping", table]); // (B)
  18. csp.go(player, ["pong", table]); // (C)
  19. yield csp.put(table, {hits: 0}); // enqueue
  20. yield csp.timeout(1000); // wait
  21. table.close();
  22. });

player defines a “process” that is instantiated twice (in line (B) and in line (C), via csp.go()). The processes are connected via the “channel” table, which is created in line (A) and passed to player via its second parameter. A channel is basically a queue.

How does CSP compare to async iteration?

  • The coding style is also synchronous.
  • Channels feel like a good abstraction for producing and consuming async data.
  • Making the connections between processes explicit, as channels, means that you can configure how they work (how much is buffered, when to block, etc.).
  • The abstraction “channel” works for many use cases: communication with and between web workers, distributed programming, etc.

5.7.2. Alternative 2: Reactive Programming

The following code demonstrates Reactive Programming via the JavaScript library RxJS:

  1. const button = document.querySelector('button');
  2. Rx.Observable.fromEvent(button, 'click') // (A)
  3. .throttle(1000) // at most one event per second
  4. .scan(count => count + 1, 0)
  5. .subscribe(count => console.log(`Clicked ${count} times`));

In line (A), we create a stream of click events via fromEvent(). These events are then filtered so that there is at most one event per second. Every time there is an event, scan() counts how many events there have been, so far. In the last line, we log all counts.

How does Reactive Programming compare to async iteration?

  • The coding style is not as familiar, but there are similarities to Promises.
  • On the other hand, chaining operations (such as throttle()) works well for many push-based data sources (DOM events, server-sent events, etc.).
  • Async iteration is for pull streams and single consumers. Reactive programming is for push streams and potentially multiple consumers. The former is better suited for I/O and can handle backpressure. There is an ECMAScript proposal for Reactive Programming, called “Observable” (by Jafar Husain).

5.8. Further reading