Learn · Mental model

Pipeline model

An Exstream pipeline connects a source to transformations and a terminal consumer. The consumer starts the work and controls when the next record is useful.

Three parts

source → operators → sink

A source may be an iterable, async iterable, Web ReadableStream, Node-readable stream, promise, or Exstream generator. It produces the original values.

Operators transform the flow. map() changes values, filter() removes them, mapAsync() coordinates asynchronous work, and format operators turn chunks into records or records into chunks.

A terminal consumer creates downstream demand. Examples include pipeTo(), toAsyncIterator(), drain(), and values().

Chains are lazy

const activeOrders = exstream(source)
  .map(normalizeOrder)
  .filter((order) => order.active)

This code describes a pipeline. It does not consume the source yet. Work begins when something asks activeOrders for values:

for await (const order of activeOrders.toAsyncIterator()) {
  await saveOrder(order)
}

Open “Pipeline model” in the playground

That distinction matters for files, network responses, and generators with side effects. Constructing a pipeline is not the same as running it.

Demand moves upstream

The consumer asks for capacity. Operators pass that demand toward the source. Values travel back toward the consumer. A slow destination therefore influences when the source is read again.

This is the basis of backpressure. It is also why a terminal operation should be visible in application code: it identifies who owns completion and failure.

Sync stays sync

With a synchronous source and synchronous operators, Exstream keeps a synchronous path:

const values = exstream([1, 2, 3])
  .map((value) => value * 2)
  .valuesSync()

An asynchronous source or operator changes how the result must be consumed. Use toAsyncIterator(), pipeTo(), drain(), or await the relevant terminal promise.

Next

Continue with transform data or jump to consume a pipeline if the terminal boundary is your immediate problem.