streamPromises.pipeline()
Await a chain of streams that cleans up every stage when one fails.
pipeline(source[, ...transforms], destination[, options])- Added in Node.js v15.0.0
Parameters
| Parameter | Type | Description |
|---|---|---|
$source | Stream | Iterable | AsyncIterable | Function | ReadableStream | Where the data comes from. A function stage returns a Promise or AsyncIterable. |
$...transformsoptional | Stream | Function | TransformStream | Zero or more middle stages. A function stage receives the previous stage as an AsyncIterable and returns a Promise or AsyncIterable. |
$destination | Stream | Function | WritableStream | The final stage. A function stage receives the previous stage as an AsyncIterable and returns a Promise or AsyncIterable. |
$streams | Stream[] | Iterable[] | AsyncIterable[] | Function[] | ReadableStream[] | WritableStream[] | TransformStream[] | Alternative form: an array of every stage passed as a single argument. |
$optionsoptional | Object | Pipeline options. |
$options.signaloptional | AbortSignal | Aborts the pipeline and destroys its streams when signalled, rejecting the promise with an AbortError. |
$options.endoptional | boolean= true | End the destination stream when the source stream ends. Transform streams are always ended, even if this value is false. |
Return value
PromiseA promise that fulfills when the pipeline is complete, and rejects with the first error to occur.
Errors
Rejects with the first error emitted by any stage, or with an AbortError if the signal option is aborted before the pipeline completes.
Changelog
- PHP v19.7.0Added support for web streams (also in v18.16.0).
- PHP v18.0.0Added the end option, which can be set to false to prevent automatically closing the destination stream when the source ends (also in v17.2.0 and v16.14.0).
This is pipeline() from node:stream/promises: identical semantics to the callback form — errors forwarded, every stream destroyed on failure — but it returns a promise, so a stream chain sits inside ordinary try/catch and await alongside the rest of your async code.
When to reach for it
By default, in any modern codebase. There is no reason to reach for the callback form unless you are inside a callback-shaped API already. Because the promise rejects rather than delivering an error to a callback, a failing stream chain propagates like any other exception: out of the function, into your existing error middleware, into a test's rejects assertion.
What trips people up
The promise resolving does not mean the destination is safe to read from in every case — for a file it does, since the callback waits for 'close', but for a destination you passed { end: false } the stream is still open by design.
The signal option aborts the pipeline and destroys the streams, but it cannot un-write what already reached the disk. A cancelled download leaves a truncated file behind; delete it in your catch block.
Also note end: false does not apply to Transform stages — they are always ended.
Example
When to use it
- Compress or copy a file inside an async function with try/catch error handling
- Cancel a long-running stream chain from an AbortController on request timeout
- Stream a database export through transforms in a job that awaits completion
- Assert that a broken pipeline rejects, in an async test
Frequently asked questions
How do I use await with streams in Node.js?
Import pipeline from node:stream/promises and await it; the promise fulfills when the whole chain completes and rejects with the first error. This replaces the callback form in node:stream for async code.
How do I cancel a stream pipeline?
Pass an AbortSignal as options.signal; aborting it destroys every stream and rejects the promise with an AbortError. Any data already flushed to the destination remains, so clean up partial output yourself.
What does the end option do in stream/promises pipeline?
Setting end: false stops the pipeline from closing the destination when the source ends, which lets you run several pipelines into one writable such as an HTTP response. Transform stages are always ended regardless.
More examples
Cancelling a download with an AbortSignal
Shows the abort path end to end. The signal destroys both streams and rejects with an AbortError, but the bytes already flushed are still on disk — hence the explicit rmSync in the catch. Distinguishing AbortError from a real failure matters, because a network error should propagate rather than be swallowed as a timeout.
import { pipeline } from 'node:stream/promises';
import { Readable } from 'node:stream';
import fs from 'node:fs';
const controller = new AbortController();
const timer = setTimeout(() => controller.abort(), 5000);
try {
const res = await fetch('https://example.com/large.iso');
await pipeline(
Readable.fromWeb(res.body),
fs.createWriteStream('large.iso'),
{ signal: controller.signal },
);
console.log('downloaded');
} catch (err) {
if (err.name === 'AbortError') {
// The streams were destroyed, but the partial file is still on disk.
fs.rmSync('large.iso', { force: true });
console.log('timed out, partial file removed');
} else {
throw err;
}
} finally {
clearTimeout(timer);
}Keeping an HTTP response open across two pipelines
Without end: false the first pipeline would close the response and the second would fail with ERR_STREAM_WRITE_AFTER_END. Note the catch: once a pipeline has started writing, the status line is already on the wire, so there is no way to send a 500 — destroying the socket is what signals failure to the client.
import { pipeline } from 'node:stream/promises';
import { Readable } from 'node:stream';
import http from 'node:http';
http.createServer(async (req, res) => {
res.setHeader('content-type', 'application/x-ndjson');
try {
// end: false leaves the response open for the next stage.
await pipeline(Readable.from(headerRows()), res, { end: false });
await pipeline(Readable.from(bodyRows()), res, { end: false });
res.end();
} catch (err) {
// Headers may already be sent, so destroying is the only honest option.
res.destroy(err);
}
}).listen(3000);
function* headerRows() { yield JSON.stringify({ type: 'meta' }) + '\n'; }
function* bodyRows() {
for (const id of [1, 2, 3]) yield JSON.stringify({ id }) + '\n';
}
Discussion