Record-oriented streaming for JavaScript

exstream

High-level record operations with end-to-end backpressure, bounded concurrency, reliable branching, and explicit error handling. Exstream connects async iterables, Node.js streams, and Web Streams to JavaScript transforms and destinations.

Quick start Read the documentation

Node.js 22+ Modern browsers Zero runtime dependencies

orders.pipeline.js source → transforms → destination

const orders = exstream(response.body)
  .json({ path: '$.orders[*]' })
  .mapAsync(enrichOrder, {
    concurrency: 8
  })
  .filter(isActive)

await orders
  .jsonlStringify()
  .pipeTo(destination)

inputpulled on demand

async work8 at a time

outputincremental JSONL

01 / Async processing

Set the amount of work and the output order.

mapAsync() runs promise-returning transforms with an explicit concurrency limit. Results stay in input order by default, or can leave as soon as they complete.

Async work and order →

const profiles = users.mapAsync(fetchProfile, {
  concurrency: 8,
  ordered: false,
  retry: 2,
  timeout: 5_000,
})

Concurrency: Finite and explicit
Order: Input or completion
Failure policy: Retry and timeout per attempt
Cancellation: Available through AbortSignal

02 / Backpressure

Demand travels from the destination back to the source.

When a writer slows down, transforms stop asking for more input. Exstream propagates that pressure through asynchronous operators and reliable branches instead of creating an unbounded queue between them.

Backpressure in the graph →

sourceasync iterator

transformmapAsync(8)

destinationslow writer

capacity propagates upstream

Pull sources: Pause naturally between reads
Async operators: Stop pulling when slots are full
Reliable forks: All destinations set the pace
Hot sources: Need an overflow policy

03 / Fork and merge

Build a graph without hiding its delivery rules.

fork() creates a reliable branch: every record must reach every active fork. merge() consumes streams with a limit on active inputs and an explicit order choice.

Branch and observe →

const database = source.fork()
const audit = source.fork()

await Promise.all([
  database.pipeTo(databaseWriter),
  audit.pipeTo(auditWriter),
])

const rows = pageStreams.merge(4, false)

fork(): Reliable, participates in pressure
observe(): Best-effort, bounded buffer
merge(): Limits active inner streams
Lifecycle: Terminal promises own completion

04 / Streaming operations

Join and group records without first collecting the input.

Higher-level operators cover common ETL work. On pre-sorted inputs, sortedGroupBy() retains one adjacent group and sortedJoin() performs a two-stream merge join.

Read the sortedJoin() contract →

const groups = rowsByCustomer
  .sortedGroupBy('customerId')

const joined = exstream([customers, orders])
  .sortedJoin(
    'id',
    'customerId',
    'left'
  )

Transform: map, filter, batch, reduce
Structure: split, flatten, keyBy
Ordered data: sortedGroupBy, sortedJoin
Formats: CSV, JSON, JSON Lines

05 / Error handling

Keep bad records separate from a broken pipeline.

Recoverable record errors can be replaced, skipped, or routed to a dead-letter stream. Source, destination, lifecycle, and cancellation failures remain fatal graph events and reject the terminal operation.

Errors and lifecycle →

const { output, deadLetters } =
  pipeline.routeErrors()

await Promise.all([
  output.pipeTo(destination),
  deadLetters.pipeTo(rejects),
])

errors(): Replace a failed record
skipErrors(): Drop accepted failures
routeErrors(): Split data and dead letters
failOnError(): Promote a record error to fatal

When it fits

Use Exstream when the flow has operational constraints.

If the data already fits in memory and the work is sequential, an array or a for await loop is usually simpler. Exstream becomes useful when concurrency, backpressure, multiple destinations, or error routing need to behave as one pipeline. See the tradeoffs →