pipeline() and Backpressure
Why pipeline replaced pipe, how errors and cleanup work, and async generators as stages.
readable.pipe(writable) works, but has a fatal flaw: if the readable errors, the writable is not closed, leaking file descriptors and sockets. stream.pipeline fixes this — it wires stages together, propagates errors, and destroys every stream in the chain on failure or completion.
const { pipeline } = require('node:stream/promises');
const fs = require('node:fs');
const zlib = require('node:zlib');
await pipeline(
fs.createReadStream('input.txt'),
zlib.createGzip(),
fs.createWriteStream('input.txt.gz'),
);
console.log('compressed, and cleaned up on any error');Backpressure, for free
Across a pipeline, backpressure is automatic end-to-end: a slow final writable throttles the gzip stage, which throttles the file read. Memory stays flat regardless of file size. This is the whole point of streaming and why you rarely hand-roll it.
Async generators as stages
The promise-based pipeline accepts async generator functions as stages, so you can drop plain async function* (source) { … } transforms into the chain without subclassing anything.
Example
// Stream a large DB export straight to an HTTP response: constant memory,
// gzip on the fly, cancelled cleanly if the client disconnects.
const { pipeline } = require('node:stream/promises');
const { Readable } = require('node:stream');
const zlib = require('node:zlib');
const http = require('node:http');
http.createServer(async (req, res) => {
const ac = new AbortController();
res.on('close', () => ac.abort()); // client gone -> stop the whole chain
res.writeHead(200, {
'Content-Type': 'application/x-ndjson',
'Content-Encoding': 'gzip',
});
// Pretend this yields millions of rows from a cursor.
async function* rows() {
for (let i = 0; i < 1_000_000; i++) yield { id: i, ts: Date.now() };
}
try {
await pipeline(
Readable.from(rows()),
async function* (src) { for await (const r of src) yield JSON.stringify(r) + '\n'; },
zlib.createGzip(),
res,
{ signal: ac.signal },
);
} catch (err) {
if (err.name !== 'AbortError') console.error('export failed', err);
}
}).listen(3000);When to use it
- A data migration pipeline uses `stream.pipeline` instead of `.pipe()` so that any error in any stage (read, transform, write) tears down the whole pipeline cleanly.
- A file server detects backpressure from a slow client connection and pauses the readable source so it does not buffer gigabytes of unsent data in memory.
- A log-processing job uses `stream.pipeline` with a writable that deliberately applies backpressure to throttle the upstream CSV reader to its processing rate.
More examples
stream.pipeline with error handling
Uses `stream/promises.pipeline` which cleans up all streams and propagates the first error via a rejected Promise.
const { pipeline } = require('stream/promises');
const fs = require('fs');
const zlib = require('zlib');
try {
await pipeline(
fs.createReadStream('input.csv'),
zlib.createGzip(),
fs.createWriteStream('output.csv.gz')
);
console.log('Pipeline succeeded');
} catch (err) {
console.error('Pipeline failed:', err.message);
}Manual backpressure with pause/resume
Manually implements backpressure by pausing the readable when the writable buffer is full and resuming on `drain`.
const readable = fs.createReadStream('huge.bin');
const writable = fs.createWriteStream('copy.bin');
readable.on('data', chunk => {
const ok = writable.write(chunk);
if (!ok) readable.pause(); // apply backpressure
});
writable.on('drain', () => readable.resume());
readable.on('end', () => writable.end());Detect pipeline bottlenecks
Inserts a passthrough Transform into the pipeline to log chunk sizes and detect where throughput drops.
const { Transform } = require('stream');
const meter = new Transform({
transform(chunk, enc, cb) {
process.stderr.write(`chunk ${chunk.length}B\n`);
this.push(chunk);
cb();
},
});
await pipeline(
fs.createReadStream('data.bin'),
meter, // insert a passthrough meter
fs.createWriteStream('out.bin')
);
Discussion