Skip to content

Intro

Jump to bottom
Eugene Lazutkin edited this page Jun 6, 2026 · 18 revisions

A bank of runnable examples — each is something you would otherwise hand-roll. For the ideas behind them see Concepts; for making them fast see Performance.

Each example is self-contained. chain carries the helpers used below as statics — chain.none, chain.many, chain.finalValue, chain.stop, chain.gen, chain.asStream — so a single import chain from 'stream-chain' covers them. Utilities and JSONL modules have their own import paths, shown inline.

Simple pipeline

import chain from 'stream-chain';

import fs from 'node:fs';
import zlib from 'node:zlib';
import {Transform} from 'node:stream';

// the chain works on a stream of number objects
const pipeline = chain([
  // transforms a value
  x => x * x,
  // returns several values
  x => [x - 1, x, x + 1],
  // waits for an asynchronous operation
  async x => await getTotalFromDatabaseByKey(x),
  // returns multiple values with a generator
  function* (x) {
    for (let i = x; i >= 0; --i) {
      yield i;
    }
  },
  // keeps odd values, drops even ones
  x => (x % 2 ? x : chain.none),
  // an arbitrary Transform stream (see the note below)
  new Transform({
    writableObjectMode: true,
    transform(x, _, callback) {
      callback(null, x.toString());
    }
  }),
  // compress
  zlib.createGzip()
]);
// log errors
pipeline.on('error', error => console.log(error));
// use the pipeline, and save the result to a file
source.pipe(pipeline).pipe(fs.createWriteStream('output.txt.gz'));

The Transform above is included to show that you can drop any Node stream into a chain. It is not the idiomatic way to do its small job here: a plain x => x.toString() would do the same, fuse with the neighbouring functions, and skip a stream boundary. Mix in real streams when you need one (here, zlib.createGzip()); reach for a function otherwise — see Performance — let chain() fuse your functions.

Using asynchronous generators

import chain from 'stream-chain';

import {Transform} from 'node:stream';
import fs from 'node:fs';
import zlib from 'node:zlib';

const family = chain([
  async function* (person) {
    yield person;
    // asynchronously retrieve parents
    if (person.father) {
      yield await getPersonFromDB(person.father);
    }
    if (person.mother) {
      yield await getPersonFromDB(person.mother);
    }
    // asynchronously retrieve children, if any
    for (let i = 0; i < person.children; ++i) {
      yield await getPersonFromDB(person.children[i]);
    }
  },
  new Transform({
    writableObjectMode: true,
    transform(x, _, callback) {
      callback(null, JSON.stringify(x));
    }
  }),
  zlib.createGzip(),
  fs.createWriteStream('families.json-stream.gz')
]);

people.pipe(family);

As before, the Transform is shown for variety — x => JSON.stringify(x) is the lighter, fusable equivalent. The async generator is the star: one input fans out to zero-to-many outputs, lazily (Concepts — the model).

Combined functions

A block of plain functions can be grouped in a nested array. chain() combines such a block into a single gen() segment, running its functions with no stream machinery in between — the main lever behind the library's speed (Concepts — how performance shaped this, Performance).

import chain from 'stream-chain';

const somePipe = [
  x => x * x,
  x => 2 * x
  // ... more stages
];

const otherPipe = chain.gen(
  x => x + 1,
  x => Math.sqrt(x)
  // ... more stages
);

const myPipe = chain([
  somePipe,
  otherPipe
  // ... more stages
]);

Filter out values

Returning chain.none produces nothing, dropping the value from the stream. (It does not stop the pipeline — that is chain.stop.) Below it drops all even values. chain.none is the one "drop" signal that works in every context (Concepts — the model).

import chain from 'stream-chain';

const pipeline = chain([
  [x => (x % 2 ? x : chain.none), x => x * x, x => 2 * x]
  // ... more stages
]);

// input: 1, 2, 3, 4
// output: 2, 18

Return a final value prematurely

chain.finalValue(x) is how you short-circuit a function block. A block runs its functions left to right; often a stage produces a value that means we are done with this item — an Error carrying context, or a sentinel such as null — and the remaining stages should be skipped while that value goes straight to the output. Returning chain.finalValue(x) does exactly that: the executor emits x and bypasses the rest of the block.

import chain from 'stream-chain';

const pipeline = chain([
  [
    raw => {
      const r = parse(raw);
      return r.ok ? r.value : chain.finalValue(r.error); // short-circuit on error
    },
    value => enrich(value), // runs only for successfully parsed values
    value => format(value)
  ]
]);
// a parse error skips enrich + format and is emitted unchanged

The same mechanic in miniature — odd results are wrapped, so the trailing 2 * x stage is skipped for them:

const pipeline = chain([
  [x => x * x, x => (x % 2 ? chain.finalValue(x) : x), x => 2 * x]
  // ... more stages
]);

// input: 1, 2, 3, 4
// output: 1, 8, 9, 32

chain.finalValue and chain.stop are interpreted by stream-chain's function executor — they work inside a chain's functions, gen() / fun() blocks, and functions wrapped by asStream() / asWebStream(). A foreign Transform you drop into a chain has no provision for them. chain.stop ends the pipeline: when it surfaces through a chain()-made stream, the stream is torn down cleanly (its readable side ends).

Create Transform out of functions

Sometimes all you need is to wrap a function as a Transform — a standalone, reusable stream component. It works with combined functions too. (When a component will be composed into other pipelines, prefer the function form so it fuses — see Performance — package reusable components.)

Wrap a function

import chain from 'stream-chain';

const stream = chain.asStream(x => x + 1);

Wrap and combine functions

import chain from 'stream-chain';

const stream = chain.asStream(
  chain.gen(
    x => x * x,
    x => 2 * x
  )
);

Reading and writing JSONL files

JSON Lines (one JSON value per line) is the batteries-included I/O for object streams. The parser turns bytes/text into {key, value} records (key is the line index); the stringer turns objects back into JSONL. Full surface: JSONL.

Parse a gzipped JSONL file and process it — parser() is a function pipeline, so it fuses into the chain:

import chain from 'stream-chain';
import parser from 'stream-chain/jsonl/parser.js';

import fs from 'node:fs';
import zlib from 'node:zlib';

chain([
  fs.createReadStream('input.jsonl.gz'),
  zlib.createGunzip(),
  parser(), // bytes/text -> {key, value}
  ({value}) => (value.active ? value : chain.none),
  value => console.log(value)
]);

For local-file round trips, the file-edge composites fold the file handles into the pipeline as the only I/O edges (no per-chunk stream boundaries), and are driven with pipe + drain:

import {pipe} from 'stream-chain/utils/pipe.js';
import {drain} from 'stream-chain/utils/drain.js';
import parseFile from 'stream-chain/jsonl/file/parser.js';
import stringerToFile from 'stream-chain/jsonl/file/stringer.js';

// read input.jsonl, drop the {key, value} envelope, write values back as JSONL
const c = pipe(parseFile(), r => r.value, stringerToFile('output.jsonl'));
await drain(c('input.jsonl'));

Why the composites are faster: Performance — streams at the edges.

Slicing streams

Take

Processes only the first 5 items of a stream. (See utils for the full set of slicing helpers.)

import chain from 'stream-chain';
import take from 'stream-chain/utils/take.js';

const pipeline = chain([
  take(5)
  // ... more stages
]);

Skip

Skips the first 5 items of a stream.

import chain from 'stream-chain';
import skip from 'stream-chain/utils/skip.js';

const pipeline = chain([
  skip(5)
  // ... more stages
]);

Take & skip together

Skips 5 items, then takes the next 5. takeWithSkip does both in a single pass:

import chain from 'stream-chain';
import skip from 'stream-chain/utils/skip.js';
import take from 'stream-chain/utils/take.js';
import takeWithSkip from 'stream-chain/utils/takeWithSkip.js';

const lessEfficient = chain([
  skip(5),
  take(5)
  // ... more stages
]);

const moreEfficient = chain([
  takeWithSkip(5, 5)
  // ... more stages
]);

Conditional take

Takes while a condition is true.

import chain from 'stream-chain';
import takeWhile from 'stream-chain/utils/takeWhile.js';

const pipeline = chain([
  takeWhile(item => item !== 'separator')
  // ... more stages
]);

Conditional skip

Skips while a condition is true.

import chain from 'stream-chain';
import skipWhile from 'stream-chain/utils/skipWhile.js';

const pipeline = chain([
  skipWhile(item => item !== 'separator')
  // ... more stages
]);

Conditional take & skip together

Processes data between the first two separators.

import chain from 'stream-chain';
import skip from 'stream-chain/utils/skip.js';
import skipWhile from 'stream-chain/utils/skipWhile.js';
import takeWhile from 'stream-chain/utils/takeWhile.js';

const pipeline = chain([
  skipWhile(item => item !== 'separator'),
  skip(1), // skip the separator
  takeWhile(item => item !== 'separator')
  // ... more stages
]);

Folding AKA reducing

The streaming analogue of Array#reduce(). (More in utils.)

Folding

import chain from 'stream-chain';
import fold from 'stream-chain/utils/fold.js';

const pipeline = chain([
  fold((acc, x) => acc + x, 0)
  // ... more stages
]);

// input: 1, 2, 3
// output: 6

Scanning

scan() is like fold() but outputs every intermediate accumulator value.

import chain from 'stream-chain';
import scan from 'stream-chain/utils/scan.js';

const pipeline = chain([
  scan((acc, x) => acc + x, 0)
  // ... more stages
]);

// input: 1, 2, 3
// output: 1, 3, 6

Dedicated reducer

reduceStream() is a Writable used at the end of a pipeline to accumulate items; its accumulator is exposed as a property. It can be used like fold() and scan().

import chain from 'stream-chain';
import reduceStream from 'stream-chain/utils/reduceStream.js';

const toArray = reduceStream((acc, x) => {
  acc.push(x);
  return acc;
}, []);

const pipeline = chain([
  // ... more stages
  toArray
]);

// input: 1, 2, 3
// toArray.accumulator is [1, 2, 3]

Web Streams

The same model runs on Web Streams. Import from stream-chain/web; the chain is a {readable, writable} pair instead of a Node Duplex. The stages you write are identical — only how you feed and consume the chain changes (see Performance — pick the leanest substrate).

import chain from 'stream-chain/web';

const c = chain([x => x * x, x => (x % 2 ? x : chain.none)]);

// feed a Web ReadableStream in, pipe the results to a WritableStream out
source.pipeTo(c.writable);
c.readable.pipeTo(destination);

Advanced use cases

Using many()

chain.many([...]) emits several values from one input. It collects them in an array held in memory, so keep the fan-out bounded — a handful of values per input. For a large or unbounded fan-out use a generator stage instead, which yields lazily without accumulating (Performance).

import chain from 'stream-chain';

const pipeline = chain([
  x => chain.many([x, 10 * x]),
  x => 2 * x
  // ... more stages
]);

// input: 1, 2, 3
// output: 2, 20, 4, 40, 6, 60

Using asynchronous functions

A stage may be async; the pipeline awaits it, holding backpressure meanwhile. Keep stages synchronous when they do not actually await — see Performance — keep stages synchronous.

import chain from 'stream-chain';

const pipeline = chain([
  async x => await getItemNumberFromDB(x),
  x => 2 * x
  // ... more stages
]);

🔍 Search the wiki

stream-chain 4.x

Start here

API

Transducers

Adapters

I/O & helpers

Tuning & internals

Reference

stream-chain 2.x (legacy)

Clone this wiki locally