utils

utils is a collection of useful utilities:

Slicing:
Folding AKA reducing:
Adapting:
Stream helpers:
Driving gen pipelines:
- pipe()
- drain()
Block I/O (Node-only):
- asyncBlockReader()
- asyncBlockWriter()

Node ↔ Web mapping

For utilities that depend on a stream substrate, every Node counterpart has a Web counterpart with the same contract — different return shape (Node Readable/Writable/Duplex vs. Web ReadableStream/WritableStream/{readable, writable}), same behavior:

Node	Web
readableFrom()	readableWebStreamFrom()
reduceStream()	reduceWebStream()
makeStreamPuller()	makeWebStreamPuller()

The JSONL pair lives in its own module — see jsonl for parserStream/parserWebStream and stringerStream/stringerWebStream.

For utilities without a substrate (object slicing, folding, scanning, batching, UTF-8 repartitioning, line splitting), a single module works on both sides — pick by what you compose them into.

Slicing

`utils/skip.js`

skip(n) skips n items from the beginning of a stream.

// const {skip} = require('stream-chain/utils/skip.js');
import skip from 'stream-chain/utils/skip.js';

chain([
  dataStream,
  skip(5) // skip 5 items
]);

`utils/skipWhile.js`

skipWhile(fn) skips items from the beginning of a stream until fn returns a falsy value. fn is called with a single argument: the item. As soon as fn returns false, all items will go through and the function will not be consulted anymore. fn can be an asynchronous function.

// const {skipWhile} = require('stream-chain/utils/skipWhile.js');
import skipWhile from 'stream-chain/utils/skipWhile.js';

chain([
  dataStream,
  skipWhile(item => item.foo === 'bar') // skip items until `foo` is not `bar`
]);

`utils/take.js`

take(n, finalValue) takes n items from the beginning of a stream, then returns finalValue, which defaults to none (see defs for more details). If you want to stop a stream early, use stop as the finalValue.

// const {take} = require('stream-chain/utils/take.js');
import take from 'stream-chain/utils/take.js';

chain([
  function* () {
    for (let i = 0; ; ++i) yield i;
  },
  take(5, stop) // take 5 items and stop the stream
]);

`utils/takeWhile.js`

takeWhile(fn, finalValue) takes items from the beginning of a stream until fn returns a falsy value. fn is called with a single argument: the item. As soon as fn returns false, takeWhile() will start returning finalValue, which defaults to none (see defs for more details) and fn will not be consulted anymore. fn can be an asynchronous function. If you want to stop a stream early, use stop as the finalValue.

// const {takeWhile} = require('stream-chain/utils/takeWhile.js');
import takeWhile from 'stream-chain/utils/takeWhile.js';

chain([
  function* () {
    for (let i = 0; ; ++i) yield i;
  },
  x => x * x,
  takeWhile(item => item < 55, stop) // take items less than 55 and stop the stream
]);

`utils/takeWithSkip.js`

takeWithSkip(n, skip, finalValue) skips skip items from the beginning of a stream, then takes n items. See notes on finalValue in the take() section above. skip defaults to 0.

// const {takeWithSkip} = require('stream-chain/utils/takeWithSkip.js');
import takeWithSkip from 'stream-chain/utils/takeWithSkip.js';

chain([
  function* () {
    for (let i = 0; ; ++i) yield i;
  },
  takeWithSkip(5, 2, stop) // skip 2 items, take 5 items
]);

Folding AKA reducing

`utils/fold.js`

fold(fn, initialValue) reduces a stream into a single value. fn is called with two arguments: the accumulator and the current item. The accumulator is the result of the previous call to fn or initialValue for the very first call. fn can be an asynchronous function.

// const {fold} = require('stream-chain/utils/fold.js');
import fold from 'stream-chain/utils/fold.js';

chain([
  function* () {
    for (let i = 0; i < 4; ++i) yield i;
  },
  fold((acc, item) => acc + item, 0)
]);
// produces: 6

`utils/reduce.js`

reduce() is an alias for fold().

`utils/reduceStream.js`

reduceStream() is a stream version of reduce(). Unlike reduce(), it returns a Writable stream. The function supports two signatures:

reduceStream(fn, initialValue) — this signature corresponds to reduce() above.
reduceStream(options) — options is an object used to initialize Writable with the following custom options:
- objectMode is set to true by default.
- initial is an initial value of an accumulator. Default: 0.
- reducer is a function or an asynchronous function, which takes a current accumulator value and a current item and returns a new accumulator value. Default: (acc, value) => value.

reducer is called in the context of the created stream. It takes two arguments: the current accumulator value and the current item.

The returned stream is a writable stream. It has a special property accumulator which contains the current accumulator value.

// const {reduceStream} = require('stream-chain/utils/reduceStream.js');
import reduceStream from 'stream-chain/utils/reduceStream.js';

const r = reduceStream((acc, x) => acc + x, 0);

chain([
  function* () {
    for (let i = 0; i < 4; ++i) yield i;
  },
  r
]);

// when we are done
console.log(r.accumulator); // 6

`utils/reduceWebStream.js`

reduceWebStream() is the Web Streams counterpart of reduceStream(). It returns a {writable, result, accumulator} triple: write into writable, await result for the final accumulator (resolves on clean close, rejects on abort or reducer error), or read the running value any time via the accumulator getter.

It supports the same two signatures:

reduceWebStream(fn, initialValue) — positional.
reduceWebStream(options) — options is an object with:
- reducer — a function or asynchronous function (acc, value) => newAcc. Default: (acc, value) => value.
- initial — initial accumulator value. Default: 0. Explicit null, undefined, false, '', 0 are respected (presence-check semantics).
- strategy — optional Web Streams QueuingStrategy for the writable side. writableStrategy overrides it.

reducer is called with this bound to the return object, mirroring reduceStream's this.accumulator access.

// const {reduceWebStream} = require('stream-chain/utils/reduceWebStream.js');
import reduceWebStream from 'stream-chain/utils/reduceWebStream.js';

const r = reduceWebStream((acc, x) => acc + x, 0);

// Pipe input through r.writable, then read the final accumulator:
source.pipeTo(r.writable);
console.log(await r.result); // sum of source values

// Or poll the live accumulator during the run:
console.log(r.accumulator);

`utils/scan.js`

scan(fn, initialValue) is a companion to fold(), but unlike fold(), on every incoming item it returns the current accumulator value. fn is called with two arguments: the accumulator and the current item. fn can be an asynchronous function.

// const {scan} = require('stream-chain/utils/scan.js');
import scan from 'stream-chain/utils/scan.js';

chain([
  function* () {
    for (let i = 0; i < 4; ++i) yield i;
  },
  scan((acc, item) => acc + item, 0) // produces: 0, 1, 3, 6
]);

Adapting

`utils/streamPuller.js`

makeStreamPuller(stream) wraps a Node Readable as a non-destructive async iterator. Built on Node's built-in stream.iterator({destroyOnReturn: false}) — preserves the original 'error' value (no AbortError wrapping), synthesizes Error('Premature close') if the source is destroyed without 'end', and breaks out of for await without destroying the source.

Returns an AsyncIterableIterator<T> exposing next(), return(), and [Symbol.asyncIterator](). See the dedicated makeStreamPuller() page.

import makeStreamPuller from 'stream-chain/utils/streamPuller.js';

for await (const v of makeStreamPuller(readable)) {
  if (shouldStop(v)) break; // source remains usable
}

`utils/webStreamPuller.js`

makeWebStreamPuller(stream) wraps a Web Streams ReadableStream as a non-destructive async iterator. Built on the native stream[Symbol.asyncIterator]({preventCancel: true}) plus a cancel(reason) extension — the iterator-protocol return() can't carry a cancel reason, so cancel-with-reason is a separate method.

Returns a WebStreamPuller<T> exposing next(), return(), cancel(reason?), and [Symbol.asyncIterator](). See the dedicated makeWebStreamPuller() page.

import makeWebStreamPuller from 'stream-chain/utils/webStreamPuller.js';

const puller = makeWebStreamPuller(readable);
for await (const v of puller) {
  /* ... */
}
await puller.cancel(new Error('user aborted'));

`utils/readableFrom.js`

readableFrom(options) adapts an iterable/iterator to a Readable stream.

options is an object used to initialize Readable with the following custom options:

objectMode is set to true by default.
iterable is an iterable/iterator, which will be a source of items.

// const {readableFrom} = require('stream-chain/utils/readableFrom.js');
import readableFrom from 'stream-chain/utils/readableFrom.js';

chain([
  readableFrom({iterable: [1, 2, 3]}),
  x => console.log(x) // 1, 2, 3
]);

`utils/readableWebStreamFrom.js`

readableWebStreamFrom(options) is the Web Streams counterpart of readableFrom(). It adapts an iterable, iterator, or 0-ary function to a Web Streams ReadableStream.

Argument forms:

A plain iterable / async iterable / iterator (with Symbol.iterator or Symbol.asyncIterator).
A function (sync or async) returning a value, a generator, an async generator, or a promise of any of those.
An options object with the following custom field:
- iterable — the source (function or iterable; required).
- strategy — optional Web Streams QueuingStrategy for the readable side.

Recognizes the chain protocol on the producer side: returned none / null / undefined is skipped, returned stop terminates the stream cleanly, returned many([...]) fans out, finalValue(v) is unwrapped to v. Per-item backpressure: the pump parks when desiredSize drops to 0 and resumes on pull().

// const {readableWebStreamFrom} = require('stream-chain/utils/readableWebStreamFrom.js');
import readableWebStreamFrom from 'stream-chain/utils/readableWebStreamFrom.js';

import chain from 'stream-chain/web';

const c = chain([x => x * x]);
readableWebStreamFrom([1, 2, 3]).pipeThrough(c).pipeTo(/* ... */);

For the simple "iterable → stream" case, the platform's ReadableStream.from(iterable) is enough. Reach for readableWebStreamFrom when you need the function-source / promise-resolution / chain-protocol features that the standard built-in doesn't cover.

Stream helpers

`utils/fixUtf8Stream.js`

fixUtf8Stream() is a function that emits valid UTF-8 strings from a stream of bytes by repartitioning chunks.

It solves a simple problem: streams are not aware of any encodings and can read bytes breaking multi-byte characters. It is fine when we are dealing with ASCII, but will break international characters and symbols.

Usually it goes in the pipeline before any other text processing. Note that included JSONL parsers already use it so it is not necessary to include it. See the jsonl module for more details.

// const fixUtf8Stream = require('stream-chain/utils/fixUtf8Stream.js');
import fixUtf8Stream from 'stream-chain/utils/fixUtf8Stream.js';

chain([
  fixUtf8Stream()
  // ...
]);

The decoder picks itself: TextDecoder by default — works in every runtime (Node, Bun, Deno, browser). On Node, it asynchronously upgrades to node:string_decoder (≈2–4× faster on byte chunks) without blocking module load. The upgrade is fire-and-forget: composing fixUtf8Stream() before it lands yields a TextDecoder-backed stage (still correct); composing after yields the faster one.

For workloads where the Node fast path matters from the first call, the named export whenReady() returns a Promise that resolves once the upgrade has landed (or immediately on non-Node):

import fixUtf8Stream, {whenReady} from 'stream-chain/utils/fixUtf8Stream.js';

await whenReady(); // optional -- only needed for the Node fast path on first call
chain([fixUtf8Stream() /* ... */]);

`utils/lines.js`

lines() is a function that emits lines from a stream of bytes line by line.

// const {lines} = require('stream-chain/utils/lines.js');
import lines from 'stream-chain/utils/lines.js';

chain([
  lines()
  // ...
]);

`utils/batch.js`

batch(size) is a function that emits batches of size items as arrays. The last batch may be smaller than size.

// const {batch} = require('stream-chain/utils/batch.js');
import batch from 'stream-chain/utils/batch.js';

chain([
  function* () {
    for (let i = 0; i < 10; ++i) yield i;
  },
  batch(3) // produces: [0, 1, 2], [3, 4, 5], [6, 7, 8], [9]
]);

Driving gen pipelines

The substrate-free gen() and /core chain don't flush at end-of-input on their own — that's a job for the caller. pipe() plus drain() are the standard one-shot driver pair.

`utils/pipe.js`

pipe(...stages) is a one-shot single-value driver for a gen pipeline. It returns an async generator function that, when called with a value, drives the value through the composed stages and then flushes the pipeline with none. Without that flush, sink stages whose flushable.final() does the closing work (e.g. asyncBlockWriter that must close its FileHandle) never run.

The flush runs only after the data pass completes. If a stage throws, the error propagates and the flush is skipped — a failed pipeline is not finalized, and re-driving it would just re-run the work that threw. (Likewise a stop or an early for await break ends the pipeline without a flush.) Resource-owning source stages still release their handles regardless — a generator source gets it.return() (its finally) from the executor's abort path. A resource-owning sink that needs release on abort guards its own code with try/catch, or you do so in a catch/finally around the loop; the original exception always propagates intact.

// const {pipe} = require('stream-chain/utils/pipe.js');
import pipe from 'stream-chain/utils/pipe.js';

const c = pipe(parseFile(), r => r.value, stringerToFile('out.jsonl'));
for await (const _ of c('in.jsonl')) {
} // or: await drain(c('in.jsonl'));

The returned function is single-shot per call: each invocation builds a fresh gen internally. Stateful stages (parsers, stringers, writers) should be rebuilt per pipe(...) invocation, not reused across them.

`utils/drain.js`

drain(asyncIter) awaits an async iterable and returns its last yielded value (or undefined if it yielded nothing). The standard way to run a pipe(...)(value) whose output you don't otherwise want: sink-terminated chains (e.g. ending in stringerToFile) yield nothing and resolve to undefined; single-final-value termini come through as the resolved value.

// const {drain} = require('stream-chain/utils/drain.js');
import drain from 'stream-chain/utils/drain.js';

const total = await drain(
  pipe(
    parseFile(),
    fold((acc, r) => acc + r.value.amount, 0)
  )('in.jsonl')
);

Block I/O (Node-only)

asyncBlockReader and asyncBlockWriter are the file-edge primitives behind jsonl/file/parser.js and jsonl/file/stringer.js. They are usable on their own to plug Node-only file I/O into a gen pipeline.

`utils/asyncBlockReader.js`

asyncBlockReader([options]) returns a function (path) => AsyncGenerator<string> — open the file at path, read fixed-size blocks via fileHandle.read(), decode them through StringDecoder('utf8') (which buffers multi-byte sequences split across blocks), and yield each decoded block. The handle is closed in finally, so it's released even if the consumer aborts mid-iteration or a read fails (if the close also fails, both errors surface as an AggregateError).

options recognizes:

readBlockSize (number, default: 65536) — block size in bytes.

import asyncBlockReader from 'stream-chain/utils/asyncBlockReader.js';
import {gen} from 'stream-chain/core';

const c = gen(asyncBlockReader(), lines());
for await (const line of c('input.txt')) console.log(line);

`utils/asyncBlockWriter.js`

asyncBlockWriter(path, [options]) returns a flushable function that accepts string fragments from upstream stages, accumulates them into an in-memory buffer, and writes whole blocks to path via fileHandle.write(). The flushable's final() (signaled by none propagating through the pipe) writes any remaining tail and closes the handle — so the file is closed only via the explicit flush. Use pipe(...) so the flush runs after the data pass completes and the file is closed. On an aborted pipeline (a throw, stop, or early break) pipe does not flush; close the handle from your own catch/finally if you need it released on error. The handle is released even if a write itself fails — whether a data-pass block write or the final tail write (a write+close double-fault surfaces an AggregateError).

options recognizes:

writeBlockSize (number, default: 1 MB) — buffered block size in characters before issuing a fileHandle.write().

import asyncBlockWriter from 'stream-chain/utils/asyncBlockWriter.js';
import pipe from 'stream-chain/utils/pipe.js';
import drain from 'stream-chain/utils/drain.js';

const c = pipe(async function* (xs) {
  for (const x of xs) yield x + '\n';
}, asyncBlockWriter('out.txt'));
await drain(c(['a', 'b', 'c']));

🔍 Search the wiki

stream-chain 4.x

Start here

API

chain()
defs

Transducers

gen()
fun()

Adapters

I/O & helpers

Tuning & internals

Reference

stream-chain 2.x (legacy)

Uh oh!

utils

Node ↔ Web mapping

Slicing

utils/skip.js

utils/skipWhile.js

utils/take.js

utils/takeWhile.js

utils/takeWithSkip.js

Folding AKA reducing

utils/fold.js

utils/reduce.js

utils/reduceStream.js

utils/reduceWebStream.js

utils/scan.js

Adapting

utils/streamPuller.js

utils/webStreamPuller.js

utils/readableFrom.js

utils/readableWebStreamFrom.js

Stream helpers

utils/fixUtf8Stream.js

utils/lines.js

utils/batch.js

Driving gen pipelines

utils/pipe.js

utils/drain.js

Block I/O (Node-only)

utils/asyncBlockReader.js

utils/asyncBlockWriter.js

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

stream-chain 4.x

Clone this wiki locally

`utils/skip.js`

`utils/skipWhile.js`

`utils/take.js`

`utils/takeWhile.js`

`utils/takeWithSkip.js`

`utils/fold.js`

`utils/reduce.js`

`utils/reduceStream.js`

`utils/reduceWebStream.js`

`utils/scan.js`

`utils/streamPuller.js`

`utils/webStreamPuller.js`

`utils/readableFrom.js`

`utils/readableWebStreamFrom.js`

`utils/fixUtf8Stream.js`

`utils/lines.js`

`utils/batch.js`

`utils/pipe.js`

`utils/drain.js`

`utils/asyncBlockReader.js`

`utils/asyncBlockWriter.js`