API · Consume

`pipeTo()`

Send a stream to a reusable Exstream destination or a platform writable and await complete processing.

Example

import { createWriteStream } from 'node:fs'

await pipeline.jsonlStringify().pipeTo(createWriteStream('./orders.jsonl'))

For application-level writers, the destination can be another Exstream definition:

const ordersApi = exstream
  .pipeline()
  .batch(200)
  .mapAsync(postOrders, { concurrency: 4, ordered: false })
  .drain()

await orders.pipeTo(ordersApi)

Parameters

destination

Type Destination<T> | NodeWritableLike<T> | WritableStream<T>Required

An Exstream destination, a Node-style writable exposing write, end, events, and completion, or a WHATWG WritableStream. Other values reject the returned promise.

end

Type booleanDefault true

For Node and Web writables, set to false to leave the destination open after normal source completion. Standard output streams are never closed by Exstream.

preventClose

Type booleanDefault false

For Node and Web writables, this also leaves the destination open after successful transfer. Either end: false or preventClose: true prevents closing.

preventAbort

Type booleanDefault false

For Node and Web writables, a true value leaves the destination open after a source failure or cancellation.

signal

Type AbortSignalDefault undefined

Cancels the transfer and its source branch with the signal's reason. For a Node or Web writable, preventAbort can leave the writable open.

Passing null or undefined as options applies all defaults. Other non-object values and arrays reject the returned promise. The three lifecycle flags use normal JavaScript truthiness; the TypeScript API exposes booleans. signal must have a valid AbortSignal shape.

Completion

pipeTo() is terminal and supplies downstream demand. With an Exstream destination, it resolves only after that destination’s promise has settled and the source has been consumed. With a platform writable, it waits for the source to end and the writable to finish, close, or complete all accepted writes.

An Exstream destination created from pipeline().drain() participates in the same graph, so batch(), mapAsync(), retries, error policies, and upstream backpressure keep their normal behavior.

Node write() backpressure and Web writer.ready propagate upstream. A hot non-pausable source still needs an explicit source buffer and overflow policy.

Errors

Unhandled record errors, source failures, structural format failures, destination failures, premature destination completion, and cancellation reject the promise. Handle recoverable record errors before pipeTo() when transfer should continue.

exstream.errorInfo(error) preserves provenance: operator and source failures keep their original information, platform write and close failures report a sink origin, custom destination failures report the destination sink stage, and signal cancellation reports lifecycle abort. EXSTREAM_DESTINATION_CLOSED identifies a platform destination that completes before its source; EXSTREAM_DESTINATION_INCOMPLETE identifies a custom destination that resolves without consuming its input.

pipeTo() is an instance-only terminal and cannot be added to a reusable pipeline definition:

await stream.pipeTo(destination, options)

Signature

pipeTo(
  destination: Destination<T>,
  options?: { signal?: AbortSignal },
): Promise<void>

pipeTo(
  destination: NodeWritableLike<T> | WritableStream<T>,
  options?: PipeOptions,
): Promise<void>

interface PipeOptions {
  end?: boolean
  signal?: AbortSignal
  preventAbort?: boolean
  preventClose?: boolean
}

destination(), drain(), async iteration, consume a pipeline

pipeTo()