Asynchronous Iteration

Up to this point, every measurement was a one-time result: request, wait, get. A station, though, produces data continuously. Its result is represented not by a single promise but by an open-ended sequence — and this sequence’s elements arrive over time, one at a time.

This lesson takes up that sequence. The ordinary iteration protocol falls short here: the next element’s readiness has to be waited for, yet a next call is required to return its value immediately. The solution is to extend the protocol so it returns a promise.

The Asynchronous Iterator Protocol

The iteration protocol introduced in the Objects and Functions in JavaScript course expects a next call to return an object carrying value and done fields. In the asynchronous protocol, the only difference is that the next call returns a promise that fulfills with such an object.

An object implementing the protocol defines a method named Symbol.asyncIterator instead of Symbol.iterator.

const SOURCE = {
  A1: { delay: 30, value: 21.4 },
  B2: { delay: 10, value: 19.8 },
  C3: { delay: 20, value: 23.1 },
  D4: { delay: 15, error: "sensor fault" },
  E5: { delay: 5, error: "no connection" },
};

function requestMeasurement(station) {
  return new Promise((resolve, reject) => {
    const record = SOURCE[station];
    setTimeout(() => {
      if (record.error) reject(new Error(`${station}: ${record.error}`));
      else resolve({ station, value: record.value });
    }, record.delay);
  });
}

const measurementSource = {
  stations: ["A1", "B2"],
  [Symbol.asyncIterator]() {
    const stations = this.stations;
    let index = 0;
    return {
      async next() {
        if (index >= stations.length) return { value: undefined, done: true };
        const measurement = await requestMeasurement(stations[index]);
        index += 1;
        return { value: measurement, done: false };
      },
    };
  },
};

for await (const measurement of measurementSource) {
  console.log("from the protocol:", measurement.station, measurement.value);
}

from the protocol: A1 21.4
from the protocol: B2 19.8

The for await loop calls next on every turn, awaits the returned promise, and continues until the done field is true. The loop’s body can be asynchronous too; each turn waits for the previous one to finish.

Asynchronous Generators

Writing the protocol by hand means carrying state variables by hand. An asynchronous generator takes over this job: it is defined as async function*, and its body can use both await and yield.

const SOURCE = {
  A1: { delay: 30, value: 21.4 },
  B2: { delay: 10, value: 19.8 },
  C3: { delay: 20, value: 23.1 },
  D4: { delay: 15, error: "sensor fault" },
  E5: { delay: 5, error: "no connection" },
};

function requestMeasurement(station) {
  return new Promise((resolve, reject) => {
    const record = SOURCE[station];
    setTimeout(() => {
      if (record.error) reject(new Error(`${station}: ${record.error}`));
      else resolve({ station, value: record.value });
    }, record.delay);
  });
}

async function* measurementStream(stations) {
  for (const station of stations) {
    console.log("producer: request started —", station);
    yield await requestMeasurement(station);
  }
}

for await (const measurement of measurementStream(["A1", "B2", "C3"])) {
  console.log("consumer: processed —", measurement.station, measurement.value);
}
console.log("stream finished");

producer: request started — A1
consumer: processed — A1 21.4
producer: request started — B2
consumer: processed — B2 19.8
producer: request started — C3
consumer: processed — C3 23.1
stream finished

The output’s order shows this structure’s most important trait: the producer does not request the next item until the consumer has processed the current one. The generator suspends at the yield point and continues only once a new next call arrives. This is called a pull-based stream; the consumer sets the pace.

Errors and Early Exit

A rejection inside the stream is thrown at the location of the for await loop and is caught with an ordinary try/catch. The caught error terminates the loop.

const SOURCE = {
  A1: { delay: 30, value: 21.4 },
  C3: { delay: 20, value: 23.1 },
  D4: { delay: 15, error: "sensor fault" },
};

function requestMeasurement(station) {
  return new Promise((resolve, reject) => {
    const record = SOURCE[station];
    setTimeout(() => {
      if (record.error) reject(new Error(`${station}: ${record.error}`));
      else resolve({ station, value: record.value });
    }, record.delay);
  });
}

async function* measurementStream(stations) {
  for (const station of stations) {
    yield await requestMeasurement(station);
  }
}

try {
  for await (const measurement of measurementStream(["A1", "D4", "C3"])) {
    console.log("consumer:", measurement.station, measurement.value);
  }
} catch (error) {
  console.log("stream stopped with an error:", error.message);
}

consumer: A1 21.4
stream stopped with an error: D4: sensor fault

The third station never got its turn: the generator terminated at the point where the error was thrown. If the stream is meant to continue despite the error, the error has to be caught inside the generator and converted into a result object — this is the stream counterpart of the previous lesson’s allSettled approach.

Exiting the loop with break also terminates the generator. This gives an opportunity to close opened resources: the generator’s finally block runs in either case.

const SOURCE = {
  A1: { delay: 30, value: 21.4 },
  B2: { delay: 10, value: 19.8 },
  C3: { delay: 20, value: 23.1 },
};

function requestMeasurement(station) {
  return new Promise((resolve) => {
    const record = SOURCE[station];
    setTimeout(() => resolve({ station, value: record.value }), record.delay);
  });
}

async function* measurementStream(stations) {
  try {
    for (const station of stations) {
      yield await requestMeasurement(station);
    }
  } finally {
    console.log("producer: cleanup ran");
  }
}

for await (const measurement of measurementStream(["A1", "B2", "C3"])) {
  console.log("consumer:", measurement.station);
  if (measurement.station === "B2") break;
}
console.log("exited the loop");

consumer: A1
consumer: B2
producer: cleanup ran
exited the loop

Cleanup ran before the loop was exited. In generators reading a continuous resource — an open connection, a set-up timer — this block is mandatory; otherwise the resource stays open even after the consumer gives up.

for await Over an Array of Promises

for await also works on an ordinary array that is not an asynchronous iterator: it awaits each element with await. This is the short way to consume an array of promises in order.

const SOURCE = {
  A1: { delay: 30, value: 21.4 },
  B2: { delay: 10, value: 19.8 },
  C3: { delay: 20, value: 23.1 },
};

function requestMeasurement(station) {
  return new Promise((resolve) => {
    const record = SOURCE[station];
    setTimeout(() => resolve({ station, value: record.value }), record.delay);
  });
}

const promises = ["A1", "B2", "C3"].map((station) => requestMeasurement(station));
for await (const measurement of promises) {
  console.log("consumed in order:", measurement.station, measurement.value);
}

consumed in order: A1 21.4
consumed in order: B2 19.8
consumed in order: C3 23.1

The distinction here matters: the requests started together, the results were consumed in order. Results are processed in the array’s order; A1 is awaited first even though it is the slowest. This is the desired behavior when processing order matters.

Backpressure

The concept of backpressure, introduced when queues were covered in the Data Structures course, sits at the center of asynchronous streams: if the producer is faster than the consumer, the gap between them has to pile up somewhere.

In a push-based source, the producer does not wait for the consumer; items pile up in a buffer.

const buffer = [];
let produced = 0;

const id = setInterval(() => {
  produced += 1;
  buffer.push({ order: produced });
  if (produced === 6) clearInterval(id);
}, 5);

await new Promise((resolve) => setTimeout(resolve, 45));
console.log("produced:", produced, "— waiting in buffer:", buffer.length);

produced: 6 — waiting in buffer: 6

Since the consumer never reads, all six of the six items sit waiting in memory. Had production been open-ended, the buffer would have grown without limit and exhausted memory. This is one of the leak forms covered in the memory topic.

In a pull-based stream, this problem does not exist by definition: the generator produces the next value only when it is requested, so there is nothing to pile up. Putting a push-based source behind a pull-based interface — that is, bounding the buffer and stopping production once it fills — is the basic pattern of stream design.

Summary

• The asynchronous iterator protocol is identical to the ordinary iteration protocol, except that next returns its result as a promise; the object defines Symbol.asyncIterator.
• Generators in async function* form take over the protocol’s state management and allow both await and yield in the body.
• for await waits for the next value on every turn; a rejection in the stream is thrown at the loop’s location and terminates the loop.
• On early exit, the generator’s finally block runs; cleanup for continuous resources is written there.
• In a pull-based stream, the consumer sets the pace and backpressure forms on its own; in a push-based stream, memory runs out if the buffer is not bounded.

Next Step

The measurement source used in this topic was always simulated with a timer, in the same process. The next topic moves the source outside the process: requests and responses over the network. There, the structure of request and response objects, reading a body as a stream, and this lesson’s asynchronous iteration applied directly to network bodies will be taken up.