Node.js Stream Pipeline — Complete Guide to stream.pipeline and stream.promises.pipeline
In this tutorial, you will learn about Node.js Stream Pipeline. We cover key concepts, practical examples, and best practices to help you master this topic.
Node.js stream.pipeline chains multiple streams together, automatically forwarding data, managing backpressure, propagating errors, and cleaning up resources across the entire pipeline.
What You'll Learn
By the end of this tutorial, you'll use pipeline() to compose streams, handle errors correctly, leverage the promise-based API, build reusable pipeline components, and avoid common piping pitfalls.
Why Pipeline Matters
The pipe() method does not forward errors and leaves streams in inconsistent states. pipeline() was introduced to solve these safety issues with proper cleanup.
Real-World Use
A log processing service reads gzipped log files, decompresses, parses JSON, filters error entries, and writes to a database. pipeline() ensures that any error in this chain destroys all streams properly.
Pipeline Path
flowchart LR
A[Backpressure] --> B[Pipeline]
B --> C[Buffers]
C --> D[File System]
D --> E[Error Handling]
B --> F{You Are Here}
style F fill:#f90,color:#fff
Pipeline Basics
pipeline() takes streams as arguments plus a final callback. It pipes data from first to last, forwarding errors and cleaning up.
const { pipeline } = require("node:stream");
const fs = require("node:fs");
const zlib = require("node:zlib");
const source = fs.createReadStream("input.txt.gz");
const dest = fs.createWriteStream("output.txt");
pipeline(
source,
zlib.createGunzip(),
dest,
(err) => {
if (err) {
console.error("Pipeline failed:", err.message);
} else {
console.log("Pipeline completed successfully");
}
}
);
Promise-Based Pipeline
The promise-based API returns a Promise, enabling async-await usage.
const { pipeline } = require("node:stream/promises");
const fs = require("node:fs");
const zlib = require("node:zlib");
async function decompressFile(input, output) {
try {
await pipeline(
fs.createReadStream(input),
zlib.createGunzip(),
fs.createWriteStream(output)
);
console.log("Decompression complete");
} catch (err) {
console.error("Decompression failed:", err.message);
}
}
decompressFile("data.tar.gz", "data.tar");
Pipeline with Transform
Insert Transform streams into the pipeline for data modification.
const { pipeline } = require("node:stream/promises");
const { Transform } = require("node:stream");
const fs = require("node:fs");
const upperCase = new Transform({
transform(chunk, enc, cb) {
cb(null, chunk.toString().toUpperCase());
},
});
async function processFile(input, output) {
await pipeline(
fs.createReadStream(input, "utf8"),
upperCase,
fs.createWriteStream(output)
);
}
processFile("input.txt", "output.txt").catch(console.error);
Error Propagation in Pipeline
Unlike pipe(), pipeline() propagates errors from any stream in the chain to the final callback.
const { pipeline } = require("node:stream");
const fs = require("node:fs");
const brokenStream = new (require("node:stream").Transform)({
transform(chunk, enc, cb) {
cb(new Error("Stream processing error"));
},
});
pipeline(
fs.createReadStream(__filename),
brokenStream,
fs.createWriteStream("/dev/null"),
(err) => {
console.log("Error type:", err?.constructor.name);
console.log("Error message:", err?.message);
// All streams in the pipeline are destroyed automatically
}
);
Reusable Pipeline Components
Create utility functions that return stream pipelines for common operations.
const { pipeline } = require("node:stream/promises");
const zlib = require("node:zlib");
const fs = require("node:fs");
async function compress(input, output) {
return pipeline(
fs.createReadStream(input),
zlib.createGzip({ level: 9 }),
fs.createWriteStream(output)
);
}
async function decompress(input, output) {
return pipeline(
fs.createReadStream(input),
zlib.createGunzip(),
fs.createWriteStream(output)
);
}
Common Mistakes
1. Using pipe() Instead of pipeline()
pipe() does not destroy streams on error, leading to resource leaks and hanging processes.
2. Not Awaiting the Promise Pipeline
Forgetting await causes unhandled rejections. Always await or attach .catch() to promise-based pipeline.
3. Mixing pipe and pipeline
Do not use pipe() inside a pipeline. Let pipeline manage all connections.
4. Ignoring Final Callback Error
Always check the error in the callback. Unchecked errors hide failures.
5. Not Handling Backpressure in Custom Streams
Custom streams in a pipeline must implement proper backpressure. pipeline forwards backpressure, but your _write/_transform must respect it.
Practice Questions
1. What problem does pipeline() solve that pipe() does not?
pipe() does not propagate errors or clean up streams. pipeline() destroys all streams on error and calls the callback.
2. How do you use pipeline with async-await?
Use pipeline from the promises API: const { pipeline } = require("node:stream/promises").
3. What happens to all streams in a pipeline when one fails?
pipeline() destroys all streams in the chain, closing file descriptors and freeing resources.
4. Can pipeline handle more than two streams?
Yes. pipeline() accepts any number of streams: pipeline(source, t1, t2, t3, dest, callback).
5. Challenge: Create a pipeline that counts bytes processed.
const { Transform } = require("node:stream");
class Counter extends Transform {
constructor() { super(); this.bytes = 0; }
_transform(chunk, enc, cb) {
this.bytes += chunk.length;
cb(null, chunk);
}
_flush(cb) { console.log(`Processed ${this.bytes} bytes`); cb(); }
}
FAQ
Mini Project: Log Processing Pipeline
Build a pipeline that reads, filters, and writes log files.
const { pipeline } = require("node:stream/promises");
const { Transform } = require("node:stream");
const fs = require("node:fs");
class ErrorFilter extends Transform {
constructor() { super({ objectMode: true }); }
_transform(line, enc, cb) {
if (line.toString().toLowerCase().includes("error")) {
cb(null, line);
} else {
cb();
}
}
}
class LineSplitter extends Transform {
constructor() { super({ readableObjectMode: true }); this.buffer = ""; }
_transform(chunk, enc, cb) {
this.buffer += chunk.toString();
const lines = this.buffer.split("\n");
this.buffer = lines.pop();
lines.forEach((l) => this.push(l));
cb();
}
}
What's Next
Node.js Buffers Node.js File System Node.js Error Handling
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro