Learn · Terminal operations

Consume a pipeline

The terminal consumer starts demand, owns completion, and is where unhandled pipeline failures become visible.

Write to a stream

await pipeline.pipeTo(destination)

Open “Consume a pipeline” in the playground

Use pipeTo() for a Node-style writable or Web WritableStream. Its promise settles only when the transfer completes. It rejects on an unhandled record error, source or destination failure, structural format error, or cancellation.

Pull with async iteration

for await (const record of pipeline.toAsyncIterator()) {
  await writeRecord(record)
}

The loop body is the destination. Awaiting it before the next iteration naturally keeps the consumer from running ahead.

Pass a signal when the caller owns cancellation:

for await (const record of pipeline.toAsyncIterator({ signal })) {
  await writeRecord(record)
}

Run side effects

await pipeline.tap(sendMetric).drain()

drain() supplies demand and discards output. Use it for a side-effecting pipeline with no writer. Unlike start(), it waits until completion or failure.

Collect deliberately

const records = await pipeline.toPromise()

Collection keeps the complete result in memory. It is appropriate only when the output is known to fit. For a large or unbounded flow, use a streaming destination instead.

Name ownership

Application code should make the terminal operation easy to find. That line determines who waits, catches fatal errors, and cancels early. A pipeline without an intentional terminal owner is only a description.