Stream pipeline()
Overview
The .pipe() method is famous, but it has a horrific architectural flaw: Error Handling. If you chain read.pipe(transform).pipe(write), and the middle transform stream throws an error (e.g., corrupt GZIP data), the error does NOT propagate. The other streams stay open forever, permanently leaking memory. To fix this, Node introduced the pipeline() utility. It safely chains streams together, manages backpressure, and guarantees that if ANY stream in the chain fails, EVERY stream is instantly and safely destroyed to prevent memory leaks.
Syntax
const fs = require('fs');
const zlib = require('zlib');
// Import the modern Promise-based pipeline utility!
const { pipeline } = require('stream/promises');
async function safelyCompressFile() {
try {
const source = fs.createReadStream('video.mp4');
const compress = zlib.createGzip();
const destination = fs.createWriteStream('video.mp4.gz');
// Instead of source.pipe(compress).pipe(destination)...
// We use pipeline()! It safely manages the entire chain.
await pipeline(
source,
compress,
destination
);
console.log("Pipeline finished perfectly!");
} catch (err) {
// If ANY of the three streams fail, this catches it,
// and safely cleans up all memory leaks automatically!
console.error("Pipeline crashed:", err);
}
}Common Pitfalls
- Still using
.pipe()in modern production code. The Node.js core team strongly advises against using.pipe()for production data flows precisely because you have to manually attach an.on('error')listener to every single stream in the chain and manually call.destroy()on the others. Always use the Promise-basedpipeline(). - Misordering the streams in the arguments. The
pipeline(A, B, C)strictly flows data from Left to Right. If you put the Writable stream first, the pipeline instantly crashes because you cannot extract data from a Writable stream.
Interview Questions
pipeline() utility solve over the traditional .pipe() method?It solves the 'Dangling Stream' memory leak. If a .pipe() chain throws an error, the source stream doesn't know, and continues pumping data into a broken pipe, eating RAM. pipeline() automatically propagates the error and violently destroys every stream in the array.
Real-World Example
Using pipeline to safely proxy a massive file download directly from an AWS S3 bucket to the user's browser, without ever saving it to the server's hard drive.
const { pipeline } = require('stream/promises');
app.get('/download', async (req, res) => {
try {
// A Readable stream coming directly from the Internet (AWS)
const awsStream = await s3.getObject({ Bucket: 'b', Key: 'video.mp4' }).createReadStream();
// Pipe it directly into the Express Response object (which is a Writable stream!)
await pipeline(awsStream, res);
} catch (err) {
res.status(500).send("Download failed.");
}
});Check Your Knowledge
Test your understanding of Stream pipeline() with these quick questions.