API · Compose

`through()`

Attach a reusable transformation to the current stream.

Example

const normalizeOrder = exstream
  .pipeline()
  .map((order) => ({ ...order, total: Number(order.total) }))
  .filter((order) => order.total > 0)

const normalized = exstream(rows).through(normalizeOrder)

Parameters

target

Type Pipeline | function | Node Transform

A reusable pipeline is instantiated for this attachment. A function receives the current Exstream and returns the transformed Exstream. In Node.js, a native duplex or transform is also accepted and its readable side becomes the returned Exstream.

Composition

A function can package a small transformation without changing its behavior:

const activeOnly = (stream) => stream.filter((order) => order.active)
const active = orders.through(activeOnly)

A pipeline is a reusable definition. Every attachment creates an independent operator chain, so buffers and other operator state are not shared:

const normalize = exstream.pipeline().map(normalizeOrder)

const apiOrders = exstream(apiRows).through(normalize)
const fileOrders = exstream(csvRows).through(normalize)

An empty pipeline is an explicit identity target. Exstream recognizes it before instantiation and returns the current stream without adding a node, queue, or per-record work. This makes conditional composition cheap:

const transform = shouldNormalize ? exstream.pipeline().map(normalizeOrder) : exstream.pipeline()

const orders = source.through(transform)

In Node.js, pass a duplex or transform when data must continue through its readable side:

const decompressed = exstream(compressedInput).through(zlib.createGunzip())

Use pipeTo() for a write-only terminal destination:

await orders.pipeTo(createWriteStream('orders.jsonl'))

The target determines output type, ordering, buffering, and concurrency. Backpressure, errors, and cancellation follow the connected graph; through() does not add its own queue or error boundary.

Errors

Passing null, undefined, a live Exstream, a Node writable-only stream, or a second options argument throws when attached. A transform function that throws also escapes the attachment call. Once connected, record errors and fatal failures follow the resulting graph.

Forms

Reusable pipeline definitions also expose through() for appending another pipeline or transform function:

stream.through(reusablePipeline)
stream.through((input) => input.filter(predicate))
exstream.pipeline().through(reusablePipeline)

Signature

through<U>(
  target: Pipeline<T, U> | ((stream: Exstream<T, C>) => Exstream<U>),
): Exstream<U>

through<U>(target: NodeTransformLike<T, U>): Exstream<U, C>

Composition, pipeline(), pipeTo(), toNodeTransform()

through()