42 Asynchronous iteration

42.1 Basic asynchronous iteration

42.1.1 Protocol: async iteration

To understand how asynchronous iteration works, let’s first revisit synchronous iteration. It comprises the following interfaces:

interface Iterable<T> {
  [Symbol.iterator]() : Iterator<T>;
}
interface Iterator<T> {
  next() : IteratorResult<T>;
}
interface IteratorResult<T> {
  value: T;
  done: boolean;
}

• An Iterable is a data structure whose contents can be accessed via iteration. It is a factory for iterators.
• An Iterator is a factory for iteration results that we retrieve by calling the method .next().
• Each IterationResult contains the iterated .value and a boolean .done that is true after the last element and false before.

For the protocol for asynchronous iteration, we only want to change one thing: the values produced by .next() should be delivered asynchronously. There are two conceivable options:

• The .value could contain a Promise<T>.
• .next() could return Promise<IteratorResult<T>>.

In other words, the question is whether to wrap just values or whole iterator results in Promises.

It has to be the latter because when .next() returns a result, it starts an asynchronous computation. Whether or not that computation produces a value or signals the end of the iteration can only be determined after it is finished. Therefore, both .done and .value need to be wrapped in a Promise.

The interfaces for async iteration look as follows.

interface AsyncIterable<T> {
  [Symbol.asyncIterator]() : AsyncIterator<T>;
}
interface AsyncIterator<T> {
  next() : Promise<IteratorResult<T>>; // (A)
}
interface IteratorResult<T> {
  value: T;
  done: boolean;
}

The only difference to the synchronous interfaces is the return type of .next() (line A).

42.1.2 Using async iteration directly

The following code uses the asynchronous iteration protocol directly:

const asyncIterable = syncToAsyncIterable(['a', 'b']); // (A)
const asyncIterator = asyncIterable[Symbol.asyncIterator]();

// Call .next() until .done is true:
asyncIterator.next() // (B)
.then(iteratorResult => {
  assert.deepEqual(
    iteratorResult,
    { value: 'a', done: false });
  return asyncIterator.next(); // (C)
})
.then(iteratorResult => {
  assert.deepEqual(
    iteratorResult,
    { value: 'b', done: false });
  return asyncIterator.next(); // (D)
})
.then(iteratorResult => {
  assert.deepEqual(
    iteratorResult,
     { value: undefined, done: true });
})
;

In line A, we create an asynchronous iterable over the value 'a' and 'b'. We’ll see an implementation of syncToAsyncIterable() later.

We call .next() in line B, line C and line D. Each time, we use .then() to unwrap the Promise and assert.deepEqual() to check the unwrapped value.

We can simplify this code if we use an async function. Now we unwrap Promises via await and the code looks almost like we are doing synchronous iteration:

async function f() {
  const asyncIterable = syncToAsyncIterable(['a', 'b']);
  const asyncIterator = asyncIterable[Symbol.asyncIterator]();
  
  // Call .next() until .done is true:
  assert.deepEqual(
    await asyncIterator.next(),
    { value: 'a', done: false });
  assert.deepEqual(
    await asyncIterator.next(),
    { value: 'b', done: false });
  assert.deepEqual(
    await asyncIterator.next(),
    { value: undefined, done: true });
}

42.1.3 Using async iteration via for-await-of

The asynchronous iteration protocol is not meant to be used directly. One of the language constructs that supports it is the for-await-of loop, which is an asynchronous version of the for-of loop. It can be used in async functions and async generators (which are introduced later in this chapter). This is an example of for-await-of in use:

for await (const x of syncToAsyncIterable(['a', 'b'])) {
  console.log(x);
}
// Output:
// 'a'
// 'b'

for-await-of is relatively flexible. In addition to asynchronous iterables, it also supports synchronous iterables:

for await (const x of ['a', 'b']) {
  console.log(x);
}
// Output:
// 'a'
// 'b'

And it supports synchronous iterables over values that are wrapped in Promises:

const arr = [Promise.resolve('a'), Promise.resolve('b')];
for await (const x of arr) {
  console.log(x);
}
// Output:
// 'a'
// 'b'

42.2 Asynchronous generators

An asynchronous generator is two things at the same time:

• An async function (input): We can use await and for-await-of to retrieve data.
• A generator that returns an asynchronous iterable (output): We can use yield and yield* to produce data.

Therefore, an asynchronous generator has:

• Input that can be:
  • synchronous (single values, sync iterables) or
  • asynchronous (Promises, async iterables).
• Output that is an asynchronous iterable.

This looks as follows:

async function* asyncGen() {
  // Input: Promises, async iterables
  const x = await somePromise;
  for await (const y of someAsyncIterable) {
    // ···
  }

  // Output
  yield someValue;
  yield* otherAsyncGen();
}

42.2.1 Example: creating an async iterable via an async generator

Let’s look at an example. The following code creates an async iterable with three numbers:

async function* yield123() {
  for (let i=1; i<=3; i++) {
    yield i;
  }
}

Does the result of yield123() conform to the async iteration protocol?

async function check() {
  const asyncIterable = yield123();
  const asyncIterator = asyncIterable[Symbol.asyncIterator]();
  assert.deepEqual(
    await asyncIterator.next(),
    { value: 1, done: false });
  assert.deepEqual(
    await asyncIterator.next(),
    { value: 2, done: false });
  assert.deepEqual(
    await asyncIterator.next(),
    { value: 3, done: false });
  assert.deepEqual(
    await asyncIterator.next(),
    { value: undefined, done: true });
}
check();

42.2.2 Example: converting a sync iterable to an async iterable

The following asynchronous generator converts a synchronous iterable to an asynchronous iterable. It implements the function syncToAsyncIterable() that we have used previously.

async function* syncToAsyncIterable(syncIterable) {
  for (const elem of syncIterable) {
    yield elem;
  }
}

Note: The input is synchronous in this case (no await is needed).

42.2.3 Example: converting an async iterable to an Array

The following function is a solution to a previous exercise. It converts an async iterable to an Array (think spreading, but for async iterables instead of sync iterables).

async function asyncIterableToArray(asyncIterable) {
  const result = [];
  for await (const value of asyncIterable) {
    result.push(value);
  }
  return result;
}

Note that we can’t use an async generator in this case: We get our input via for-await-of and return an Array wrapped in a Promise. The latter requirement rules out async generators.

This is a test for asyncIterableToArray():

async function* createAsyncIterable() {
  yield 'a';
  yield 'b';
}
const asyncIterable = createAsyncIterable();
assert.deepEqual(
  await asyncIterableToArray(asyncIterable), // (A)
  ['a', 'b']
);

Note the await in line A, which is needed to unwrap the Promise returned by asyncIterableToArray(). In order for await to work, this code fragment must be run inside an async function.

42.2.4 Example: transforming an async iterable

Let’s implement an async generator that produces a new async iterable by transforming an existing async iterable.

async function* timesTwo(asyncNumbers) {
  for await (const x of asyncNumbers) {
    yield x * 2;
  }
}

To test this function, we use asyncIterableToArray() from the previous section.

async function* createAsyncIterable() {
  for (let i=1; i<=3; i++) {
    yield i;
  }
}
assert.deepEqual(
  await asyncIterableToArray(timesTwo(createAsyncIterable())),
  [2, 4, 6]
);

42.2.5 Example: mapping over asynchronous iterables

As a reminder, this is how to map over synchronous iterables:

function* mapSync(iterable, func) {
  let index = 0;
  for (const x of iterable) {
    yield func(x, index);
    index++;
  }
}
const syncIterable = mapSync(['a', 'b', 'c'], s => s.repeat(3));
assert.deepEqual(
  Array.from(syncIterable),
  ['aaa', 'bbb', 'ccc']);

The asynchronous version looks as follows:

async function* mapAsync(asyncIterable, func) { // (A)
  let index = 0;
  for await (const x of asyncIterable) { // (B)
    yield func(x, index);
    index++;
  }
}

Note how similar the sync implementation and the async implementation are. The only two differences are the async in line A and the await in line B. That is comparable to going from a synchronous function to an asynchronous function – we only need to add the keyword async and the occasional await.

To test mapAsync(), we use the helper function asyncIterableToArray() (shown earlier in this chapter):

async function* createAsyncIterable() {
  yield 'a';
  yield 'b';
}
const mapped = mapAsync(
  createAsyncIterable(), s => s.repeat(3));
assert.deepEqual(
  await asyncIterableToArray(mapped), // (A)
  ['aaa', 'bbb']);

Once again, we await to unwrap a Promise (line A) and this code fragment must run inside an async function.

42.3 Async iteration over Node.js streams

42.3.1 Node.js streams: async via callbacks (push)

Traditionally, reading asynchronously from Node.js streams is done via callbacks:

function main(inputFilePath) {
  const readStream = fs.createReadStream(inputFilePath,
    { encoding: 'utf8', highWaterMark: 1024 });
  readStream.on('data', (chunk) => {
    console.log('>>> '+chunk);
  });
  readStream.on('end', () => {
    console.log('### DONE ###');
  });
}

That is, the stream is in control and pushes data to the reader.

42.3.2 Node.js streams: async via async iteration (pull)

Starting with Node.js 10, we can also use asynchronous iteration to read from streams:

async function main(inputFilePath) {
  const readStream = fs.createReadStream(inputFilePath,
    { encoding: 'utf8', highWaterMark: 1024 });

  for await (const chunk of readStream) {
    console.log('>>> '+chunk);
  }
  console.log('### DONE ###');
}

This time, the reader is in control and pulls data from the stream.

42.3.3 Example: from chunks to lines

Node.js streams iterate over chunks (arbitrarily long pieces) of data. The following asynchronous generator converts an async iterable over chunks to an async iterable over lines:

/**
 * Parameter: async iterable of chunks (strings)
 * Result: async iterable of lines (incl. newlines)
 */
async function* chunksToLines(chunksAsync) {
  let previous = '';
  for await (const chunk of chunksAsync) { // input
    previous += chunk;
    let eolIndex;
    while ((eolIndex = previous.indexOf('\n')) >= 0) {
      // line includes the EOL (Windows '\r\n' or Unix '\n')
      const line = previous.slice(0, eolIndex+1);
      yield line; // output
      previous = previous.slice(eolIndex+1);
    }
  }
  if (previous.length > 0) {
    yield previous;
  }
}

Let’s apply chunksToLines() to an async iterable over chunks (as produced by chunkIterable()):

async function* chunkIterable() {
  yield 'First\nSec';
  yield 'ond\nThird\nF';
  yield 'ourth';
}
const linesIterable = chunksToLines(chunkIterable());
assert.deepEqual(
  await asyncIterableToArray(linesIterable),
  [
    'First\n',
    'Second\n',
    'Third\n',
    'Fourth',
  ]);

Now that we have an asynchronous iterable over lines, we can use the solution of a previous exercise, numberLines(), to number those lines:

async function* numberLines(linesAsync) {
  let lineNumber = 1;
  for await (const line of linesAsync) {
    yield lineNumber + ': ' + line;
    lineNumber++;
  }
}
const numberedLines = numberLines(chunksToLines(chunkIterable()));
assert.deepEqual(
  await asyncIterableToArray(numberedLines),
  [
    '1: First\n',
    '2: Second\n',
    '3: Third\n',
    '4: Fourth',
  ]);