asyncerator

Copyright © 2021–2026 Check Digit, LLC

Introduction

The asyncerator module provides three central capabilities:

• A shared Asyncerator<T> interface for async iteration across sources, operators and sinks.
• A typed wrapper around Node's stream.pipeline that returns a promise or readable stream depending on the sink.
• A library of simple source, transform and sink components that can be used standalone, or with pipeline.

Installing

npm install asyncerator

Asyncerator<T> interface

export interface Asyncerator<T> {
  [Symbol.asyncIterator](): AsyncIterableIterator<T>;
}

Asyncerator<T> describes an object that can be consumed with for await...of and whose [Symbol.asyncIterator]() method returns an iterator that is itself async iterable. The object does not need its own next() method, so both readable streams and iterator objects can satisfy this interface.

The following all implement the Asyncerator interface:

• AsyncIterableIterator<T>
• AsyncGenerator objects returned by async generator functions
• Node readable streams, including stream.Readable and streams returned by fs.createReadStream()
• readline.Interface instances
• Node's WHATWG ReadableStream

Why retain this interface?

With current Node typings and TypeScript 6, Node readable streams are compatible with the standard AsyncIterable<T> interface, which works directly with for await...of. Asyncerator remains part of this library's public API to preserve its existing typing guarantees:

• Compared with AsyncIterable<T>, it guarantees that the returned iterator can also be used with for await...of.
• Compared with AsyncIterableIterator<T>, it does not require the outer object to have a next() method.

asyncerator.pipeline

asyncerator.pipeline wraps Node's callback-based stream.pipeline. Node also provides pipeline from node:stream/promises, which supports async iterables and always returns a promise. This library provides overloads connecting its operators and chooses its return type according to the sink:

| Sink | Return type | | ------------------------------------------------------- | -------------------------------------------- | | Writable stream without a readable side | Promise<void> | | Async function returning Promise<T> | Promise<T>, resolving to the sink's result | | Async generator function or Duplex/Transform stream | Readable |

The overloads carry element types between function stages and infer the sink result through up to ten transforms. This catches incompatible function stages at compile time; Node stream stages do not provide the same element-type guarantees because their chunk types are not generic.

Quick example

import { filter, map, pipeline, split, toArray } from 'asyncerator';
import fs from 'fs';
import zlib from 'zlib';

// ...

// write a Gzipped file containing "hello" and "world" on two separate lines
await pipeline(
  ['hello', 'world'],
  map((string) => `${string}\n`),
  zlib.createGzip(),
  fs.createWriteStream(temporaryFile),
);

// read file using a pipeline
const result = await pipeline(
  fs.createReadStream(temporaryFile),
  zlib.createUnzip(),
  split('\n'),
  filter((string) => string !== ''),
  toArray,
); // ['hello', 'world']

// read file without pipeline (painful, don't do this!)
const result2 = await toArray(
  filter((string) => string !== '')(
    split('\n')(fs.createReadStream(temporaryFile).pipe(zlib.createUnzip())),
  ),
); // ['hello', 'world']

Sources

Sources return an object of type Asyncerator<T> that can be passed to pipeline as the first argument. Other than these built-in functions, the other objects that are considered a "source" are string, arrays or basically anything that implements the Readable, Iterable and AsyncIterable interfaces.

Some built-in source functions take an argument of type Asyncable<T>. Asyncables are anything that can be turned into an Asyncerator: normal iterators and iterables, AsyncIterators, AsyncIterables, AsyncGenerators, AsyncIterableIterators, and of course Asyncerators themselves.

all<T>(promises: Iterable<Promise<T>>): Asyncerator<T>

Similar to Promise.all(), but instead returns values as they become available via an Asyncerator. Note: the output order is not the same as the input order, the fastest promise to resolve will be first, the slowest last.

merge<T>(...iterators: Asyncable<T | Asyncable<T>>[]): Asyncerator<T>

Merge multiple asyncables into a single Asyncerator. If an iterator yields another Asyncerator, merge its output into the stream.

series<T>(...iterators: Asyncable<T>[]): Asyncerator<T>

Combine the output of iterators in a series. Requires all the iterators to complete.

Sinks

These built-in convenience sink functions all return a Promise that will wait until the Asyncerator is done before resolving, so be careful using this with endless iterators (in other words, don't do that). They can be used as the last argument to pipeline, which causes it to return a Promise, along with:

• Writable (pipeline returns a Promise<void>)
• (input: Asyncerator<Source>) => Promise<Sink> (pipeline returns a Promise<Sink>)
• Duplex (pipeline returns a Readable stream)
• ((input: Asyncerator<Source>) => AsyncIterable<Sink>) (pipeline returns a Readable stream)

reduce<Input, Output>(reduceFunction: (previousValue: Output, currentValue: Input, currentIndex: number) => Output, initialValue?: Input | Output): Promise<Input | Output | undefined>

Calls the specified callback function for all the elements in a stream. The return value of the callback function is the accumulated result, and is provided as an argument in the next call to the callback function.

Equivalent to the Javascript Array.reduce() method.

toArray<T>(iterator: Asyncable<T>): Promise<T[]>

Turns a completed Asyncerator into an Array.

toNull<T>(iterator: Asyncable<T>): Promise<void>

Drop the results of an Asyncerator into /dev/null.

toString<T>(iterator: Asyncable<T>): Promise<string>

Turns a completed Asyncerator into a string.

Operators

Operators are transformations that can be included anywhere in a pipeline except as a source. They take a stream of inputs, and generate a stream of outputs. An operator function is equivalent to an implementation of the Node Duplex or Transform interfaces, and can be used interchangeably within a pipeline.

The built-in operators all return something of type Operator<Input, Output> which is a function with the signature (input: Asyncerator<Input>) => Asyncerator<Output>.

Async generator functions that take a single Asyncerator parameter are compatible with this signature. Operators should generally be implemented using the following pattern:

function map<Input, Output>(
  mapFunction: (value: Input) => Output,
): Operator<Input, Output> {
  return async function* (iterator: Asyncerator<Input>) {
    for await (const item of iterator) {
      yield mapFunction(item);
    }
  };
}

It is straightforward to create custom operators and mix with streams, but it is important to note how they relate to streams:

• returning (exiting) out of the function is equivalent to the end event of a stream. The pipeline will complete.
• throw-ing an error is equivalent to the error event of a stream. The pipeline will throw the error.
• streams in object mode will swallow undefined values emitted by sources and prior operators.
• streams in object mode will error on null values emitted by sources and prior operators.
• non-object mode streams will error on any value that is not a string, Buffer or Uint8Array.

after<Input>(value: Input): Operator<Input, Input>

Emit a value after stream completes. Useful for adding a footer to a stream.

before<Input>(value: Input): Operator<Input, Input>

Emit a value before a stream starts. Useful for adding a header to a stream.

filter<Input>(filterFunction: (value: Input, index: number) => boolean): Operator<Input, Input>

Similar to Array.filter, only emit values from input for which filterFunction returns true.

flat<Input>(depth = 1): Operator<Input, Input extends (infer T)[] ? T : Input>

Similar to Array.flat, flatten array inputs into a single sequence of values.

forEach<Input>(forEachFunction: (value: Input, index: number) => void): Operator<Input, Input>

Similar to Array.forEach, call forEachFunction for each value in the stream.

map<Input, Output>(mapFunction: (value: Input, index: number) => Output): Operator<Input, Output>

Similar to Array.map, transform each value using mapFunction.

race<Input, Output>(raceFunction: (value: Input) => Promise<Output>, concurrent?: number): Operator<Input, Output>

Apply stream of values to the raceFunction, emitting output values in order of completion. By default, allows up to 128 concurrent values to be processed.

sequence<Input>(sequenceFunction: (index: number) => Promise<Input>): Operator<Input, Input>

The sequenceFunction will be called repeatedly with an incrementing numerical parameter, returning a Promise that resolves with the same type as Input and is inserted into the stream. The sequence operator passes through all other values. Because the sequenceFunction returns a Promise, it can delay its response (using setTimeout) to emit values on a regular schedule, e.g., once a second:

pipeline(
  ...
  sequence(async () => {
    await new Promise((resolve) => {
      setTimeout(resolve, 1000);
    });
    return 'this value is emitted once a second!';
  })
  ...
);

skip<Input>(numberToSkip = 1): Operator<Input, Input>

Skip numberToSkip values at the start of a stream.

split<Input extends { toString: () => string }>(separator: string, limit?: number): Operator<Input, string>

Equivalent of the Javascript Array.split method.

Extended examples

File system

Read a CSV file, output a new CSV file based on some logic:

import { filter, map, pipeline, race, split } from 'asyncerator';
import fs from 'fs';
import timeout from '@checkdigit/timeout';
import retry from '@checkdigit/retry';

async function main() {
  await pipeline(
    fs.createReadStream('./input.csv'),
    // turn buffers into string chunks
    map((buffer: Buffer) => buffer.toString()),
    // split chunks into lines
    split('\n'),
    // remove empty lines, and CSV header line
    filter(
      (string) =>
        string !== '' && string !== '"header1","header2","header3","header4"',
    ),
    // transform string into an object
    map((line: string) => ({
      field1: line.split(',')[0] as string,
      field4: BigInt(line.split(',')[3] as string),
    })),
    // perform concurrent requests (up to 128 by default) - Note: DOES NOT PRESERVE ORDER
    race(
      retry((item) =>
        timeout(
          (async ({ field1, field4 }) => ({
            calculated: await someAsyncNetworkAPIFunction(field1),
            field1,
            field4, // type is infered to be a BigInt, because Typescript is awesome
          }))(item),
        ),
      ),
    ),
    // demonstrate use of an async generator to filter and transform objects into a string
    // (FYI could be done more easily using map and filter)
    async function* (iterator) {
      for await (const item of iterator) {
        if (doWeOutput(item)) {
          yield `${item.field1},${item.field2},${item.calculated}\n`;
        }
      }
    },
    fs.createWriteStream('./output.csv'),
  );
}

Note this code, in addition to the built-in asyncerator functionality, also uses @checkdigit/timeout and @checkdigit/retry to implement retries with timeouts and exponential back-off.

Sockets

A pipeline is an effective method for managing socket connections. On the server side, the inbound socket can be both a source and a sink. On the client, a connection can be treated as a transform that takes an inbound stream and provides the output stream (from the server) to the next stage of the pipeline:

import net from 'net';
import { filter, pipeline, map, split, toArray } from 'asyncerator';

// ...

// echo server
const server = net
  .createServer((socket) =>
    pipeline(
      socket,
      split('\n'),
      map((command) => `echo:${command}\n`),
      socket,
    ),
  )
  .listen(1234, '127.0.0.1');

// echo client
const received = await pipeline(
  'Hello Mr Server!\nRegards, Client.\n',
  new net.Socket().connect(1234, '127.0.0.1'),
  split('\n'),
  filter((line) => line !== ''),
  toArray,
); // ['echo:Hello Mr Server!', 'echo:Regards, Client.']

License

MIT