exstream

Start here · 5 minutes

Build one useful pipeline

Read records, transform them while they flow, and write the result without collecting the whole input first.

Install

For Node.js, install the package. Exstream requires Node.js 22 or newer.

npm install exstream.js

In a browser project, install the same package through your bundler. The default import selects the portable browser runtime automatically.

Choose your runtime

One pipeline, two environments.

import { createReadStream, createWriteStream } from 'node:fs'
import exstream from 'exstream.js'

await exstream(createReadStream('./orders.csv'))
  .csv({ header: true })
  .map((order) => ({
    ...order,
    total: Number(order.total),
  }))
  .filter((order) => order.status === 'active')
  .jsonlStringify()
  .pipeTo(createWriteStream('./active-orders.jsonl'))

The source and destination change with the runtime. Parsing, transformation, and flow control do not.

What the pipeline does

The example connects a real source and destination, parses CSV incrementally, converts total to a number, keeps active orders, and serializes each result as JSON Lines.

Nothing runs just because the chain exists. pipeTo() is the terminal operation: it starts demand and settles only after the destination finishes.

MemoryNo complete-file collection
FlowThe destination sets the pace
TransformOne record at a time
CompletionExplicit terminal promise

Try one change

The result follows the pipeline.

Change a transformation or output format. The bundled browser runtime runs it here.

Add bounded asynchronous work

When a record needs I/O, use mapAsync() and state the contract you need:

const enriched = orders.mapAsync(loadCustomer, {
  concurrency: 8,
  ordered: true,
  retry: 2,
  timeout: 5_000,
})

At most eight calls are active, results preserve input order, and cancelled work receives an AbortSignal through the record context.

Handle the terminal failure

try {
  await pipeline.pipeTo(destination)
} catch (error) {
  const { origin, stage } = exstream.errorInfo(error)
  console.error(`Pipeline failed in ${origin}:${stage ?? 'unknown'}`, error)
}

Recoverable record errors and fatal graph failures are separate policies. The quick start stops at the terminal boundary; the error guide explains routing, skipping, and promotion.

Continue from here