Streams Deep: Readable, Writable & Transform
Streams Deep Dive
Node.js streams come in four flavours: Readable (source), Writable (sink), Duplex (both), and Transform (duplex that mutates data in flight). Understanding how to create custom stream classes unlocks memory-efficient data processing at any scale.
Custom Readable Stream
const { Readable } = require("stream");
class CounterStream extends Readable {
constructor(to) {
super({ objectMode: true });
this.current = 1;
this.to = to;
}
_read() {
if (this.current > this.to) {
this.push(null); // Signal end-of-stream
} else {
this.push(this.current++);
}
}
}
const counter = new CounterStream(5);
counter.on("data", (n) => console.log("tick:", n));
counter.on("end", () => console.log("done"));
// tick: 1 tick: 2 tick: 3 tick: 4 tick: 5 doneCustom Transform Stream
A Transform stream sits between a Readable and Writable. Implement _transform(chunk, encoding, callback) to modify or filter each chunk, then call callback(null, transformedChunk) to push it downstream.
const { Transform } = require("stream");
class UpperCaseTransform extends Transform {
_transform(chunk, encoding, callback) {
callback(null, chunk.toString().toUpperCase());
}
}
const { pipeline } = require("stream/promises");
const fs = require("fs");
async function run() {
await pipeline(
fs.createReadStream("input.txt"),
new UpperCaseTransform(),
fs.createWriteStream("output.txt")
);
console.log("Transformed successfully!");
}
run().catch(console.error);Always use stream/promises pipeline() instead of .pipe(). It propagates errors from every stream in the chain and cleans up resources automatically — .pipe() silently swallows upstream errors.
Backpressure
Backpressure occurs when a Writable cannot consume data as fast as the Readable produces it. pipeline() and the pipe() method handle backpressure automatically by pausing the source when the destination's internal buffer is full. If you push data manually, check the return value of writable.write() — false means you should pause until the drain event fires.