API · Consume

`pipeTo()`

Write every successful value to a Node-style writable or Web WritableStream and await the complete transfer.

Signature

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

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

Example

import { createWriteStream } from 'node:fs'

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

Parameters

destination

Type NodeWritableLike<T> | WritableStream<T>Required

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

end

Type booleanDefault true

Set to false to leave the destination open after normal source completion. Standard output streams are never closed by Exstream.

preventClose

Type booleanDefault false

Also leaves the destination open after successful transfer. This matches Web Streams terminology; either end: false or preventClose: true prevents closing.

preventAbort

Type booleanDefault false

When true, a source failure or cancellation rejects the promise but leaves the destination open instead of destroying or aborting it.

signal

Type AbortSignalDefault undefined

Cancels the transfer with the signal's reason. Unless preventAbort is true, the destination is aborted as well.

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. It resolves with undefined only after the source has ended and the destination has finished, closed, or completed all writes. With the destination left open, it waits for write callbacks or promises rather than for close.

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 write or close failures, premature destination completion, and cancellation reject the promise. Handle recoverable record errors before pipeTo() when transfer should continue.

exstream.errorInfo(error) preserves provenance: source failures report a source origin, destination write and close failures report a sink origin, and signal cancellation reports lifecycle abort. EXSTREAM_DESTINATION_CLOSED identifies a destination that completes before its source.

Forms

pipeTo() is terminal and cannot be added to a reusable pipeline definition. It supports instance, standalone direct, and standalone curried forms:

await stream.pipeTo(destination, options)
await exstream.pipeTo(destination, stream)
await exstream.pipeTo(destination, options, stream)
await exstream.pipeTo(destination, options)(stream)

toAsyncIterator(), drain(), consume a pipeline

pipeTo()