stream.pipeline()
Connect streams end to end, forwarding errors and destroying them all on failure.
stream.pipeline(source[, ...transforms], destination, callback)- Added in Node.js v10.0.0
Parameters
| Parameter | Type | Description |
|---|---|---|
$source | Stream | Iterable | AsyncIterable | Function | ReadableStream | Where the data comes from. A function stage receives no source and returns an Iterable or AsyncIterable. |
$...transformsoptional | Stream | Function | TransformStream | Zero or more middle stages. A function stage receives the previous stage as an AsyncIterable and returns an AsyncIterable. |
$destination | Stream | Function | WritableStream | The final stage. A function stage receives the previous stage as an AsyncIterable and returns an AsyncIterable or a Promise whose value is passed to the callback. |
$streams | Stream[] | Iterable[] | AsyncIterable[] | Function[] | ReadableStream[] | WritableStream[] | TransformStream[] | Alternative form: an array of every stage, passed as a single argument before the callback. |
$callback | Function | Called when the pipeline is fully done, as callback(err, val) where err is the first error to occur and val is the resolved value of a promise returned by the destination. |
Return value
StreamThe destination stream, so it can be referenced after the call.
Errors
Throws ERR_INVALID_ARG_TYPE synchronously if the callback argument is missing or not a function; all other failures are reported through the callback.
Changelog
- PHP v19.7.0Added support for web streams (also in v18.16.0).
- PHP v18.0.0Passing an invalid callback now throws ERR_INVALID_ARG_TYPE instead of ERR_INVALID_CALLBACK.
- PHP v14.0.0pipeline(..., cb) now waits for the 'close' event before invoking the callback on streams expected to emit it, so the callback fires later than in older versions.
- PHP v13.10.0Added support for async generators as pipeline stages.
pipeline() connects a chain of streams and calls you back once, when the whole chain has finished or the first one has failed. Unlike pipe(), it forwards errors and calls destroy() on every stream in the chain when something goes wrong — skipping only those already finished. That single behaviour is why it replaced chained .pipe() calls in every serious codebase.
When to reach for it
Any multi-stage stream operation in code you have to maintain: read a file, gzip it, write it out; read a request body, parse it, insert into a database. Stages need not be streams — plain iterables and async generators are accepted, so a transform can be an async function* instead of a Transform subclass.
What trips people up
The callback is mandatory: omit it and you get a synchronous ERR_INVALID_ARG_TYPE, not a promise. If you want a promise, use the version from node:stream/promises.
Because every stream is destroyed on failure, a destination you meant to reuse — an HTTP response, a shared socket — will be torn down too. Only the promise form supports { end: false } to opt out.
Since v14 the callback waits for 'close', not just 'finish', so it fires slightly later than older code assumed.
Example
When to use it
- Compress a large file without leaking descriptors when the disk fills
- Parse an upload through several transforms and report the first failure once
- Insert an async generator stage between two streams without writing a Transform
- Replace a chain of .pipe() calls that crashes the process on source errors
Frequently asked questions
What is the difference between pipeline() and pipe() in Node.js?
pipeline() forwards errors and destroys every stream in the chain when one fails, while pipe() does neither, leaving descriptors open and letting an unhandled 'error' crash the process. Use pipeline() for anything beyond a throwaway script.
Can I use async/await with stream.pipeline?
Not with the callback version in node:stream — import pipeline from node:stream/promises instead, which returns a promise you can await. The callback form throws ERR_INVALID_ARG_TYPE if you omit the callback.
Does pipeline close the destination stream?
Yes, it ends and destroys the destination when the source finishes or any stage fails. Only the promise form accepts { end: false } to leave the destination open.
More examples
Cleanup on failure, which pipe() does not do
Prints wrote: a, then error: disk full, source destroyed: true, dest destroyed: true. The failure originated in the destination, yet the source was torn down as well — with source.pipe(dest) the source would still be open, still reading, and the error would have escaped as an uncaught exception.
const { pipeline, Readable, Writable } = require('node:stream');
const source = Readable.from(['a', 'b', 'c']);
const dest = new Writable({
objectMode: true,
write(chunk, encoding, callback) {
if (chunk === 'b') return callback(new Error('disk full'));
console.log('wrote:', chunk);
callback();
},
});
pipeline(source, dest, (err) => {
console.log('error:', err.message);
// Both streams were destroyed, including the one that did not fail.
console.log('source destroyed:', source.destroyed);
console.log('dest destroyed:', dest.destroyed);
});An async generator as a middle stage
Filters a log down to HTTP 500 lines and compresses the result, with the filtering expressed as an async generator rather than a Transform. The generator handles chunk boundaries by carrying a partial line between iterations, and pipeline() still applies backpressure — the read stream advances only as fast as gzip and the disk allow.
import { pipeline } from 'node:stream';
import fs from 'node:fs';
import { createGzip } from 'node:zlib';
pipeline(
fs.createReadStream('access.log', 'utf8'),
// No Transform subclass needed: any async generator is a valid stage.
async function* (source) {
let carry = '';
for await (const block of source) {
const lines = (carry + block).split('\n');
carry = lines.pop();
for (const line of lines) {
if (line.includes(' 500 ')) yield line + '\n';
}
}
if (carry.includes(' 500 ')) yield carry + '\n';
},
createGzip(),
fs.createWriteStream('errors.log.gz'),
(err) => console.log(err ? `failed: ${err.message}` : 'errors extracted'),
);
Discussion