File processing
Copying or transforming large files without loading them fully into memory.
Node.js · Function
pipeline
stream.pipeline() connects a source through optional transforms to a destination, forwarding data, errors, and cleanup. It also handles backpressure and has a promise-based API.
How it works
pipeline connects a source, optional transforms, and a destination, while coordinating data flow and cleanup. It forwards errors, handles backpressure, and can either invoke a completion callback or return a promise through node:stream/promises.
Syntax
stream.pipeline(source[, ...transforms], destination, callback)
stream.pipeline(streams, callback)
pipeline(source[, ...transforms], destination[, options])Parameters
| Name | Type | Description |
|---|---|---|
source | Stream | Iterable | AsyncIterable | Function | ReadableStream | Input stream, iterable, async iterable, function, or compatible readable source. |
transforms | Stream | Function | TransformStream | Zero or more intermediate transformation stages. |
destination | Stream | Function | WritableStream | Output stream, function, or compatible writable destination. |
callback | Function | Completion callback invoked when processing ends. |
Returns
Returns the destination pipeline stream for the callback API; the promise API returns a promise that fulfills on success and rejects on failure.
Key points
stream.pipeline()links streams, generators, and other supported stages while forwarding failures and managing cleanup.- It centralizes backpressure handling, reducing the need for manual flow-control error management.
- Async generators can serve as transformation stages, allowing asynchronous chunk processing between the source and destination.
- The callback is invoked after the complete pipeline finishes.
Common mistakes
- Failure cleanup destroys participating streams, with documented exceptions for readable streams that have already emitted completion or closure events.
- The promise form does not take a callback; await its returned promise and handle rejection for failures.
Example: Write iterable data to a file
import { pipeline } from 'node:stream/promises';
import { Readable } from 'node:stream';
import { createWriteStream } from 'node:fs';
await pipeline(
Readable.from(['hello\n', 'world\n']),
createWriteStream('output.txt'),
);This promise-based pipeline writes generated chunks to a file and rejects if a pipeline stage fails.
When to use it
Multi-stage stream flows
Connecting readable input, transformation logic, and writable output with centralized failure handling.
Asynchronous transformations
Using async generators to transform chunks asynchronously.
Interview questions
QWhat problem does stream.pipeline() solve beyond manually calling .pipe()?
pipeline connects streams or compatible iterables and generators in sequence, forwards errors, and performs cleanup. Its callback runs when the pipeline has completed.
Follow-up: How can callback-free code use it? Import pipeline from node:stream/promises; it returns a promise that resolves on success and rejects on failure.
QWhat inputs can a pipeline contain?
The function accepts a source, zero or more transforms, a destination, and a callback, or an array-like collection plus a callback. The source and stages may include streams, iterables, async iterables, functions, and Web Streams-compatible objects.
Follow-up: What does the callback receive? It receives an error when the pipeline fails; otherwise it is called without an error after completion.
QHow does pipeline support backpressure?
Pipeline coordinates backpressure and backpressure-related errors while forwarding data between stages. This makes it suitable for long-running or large data flows without manually wiring every stage.
Follow-up: How can asynchronous transformation be added? Use an async generator as a transform stage, yielding transformed chunks from the received source.
Related
streamstream/promisesstream.composeReadableWritableTransform