Node.js Streams Deep Dive — Complete Guide to Readable, Writable, Transform, Duplex

In this tutorial, you will learn about Node.js Streams Deep Dive. We cover key concepts, practical examples, and best practices to help you master this topic.

Node.js streams deep dive covers the four fundamental stream types: Readable (source), Writable (sink), Transform (modify), and Duplex (both), with custom implementation and backpressure management.

What You'll Learn

By the end of this tutorial, you'll create custom Readable, Writable, Transform, and Duplex streams, implement backpressure, compose pipelines with pipeline(), and choose the right type.

Why Streams Matter

Streams handle data piece by piece instead of loading everything into memory. They enable processing large files, network responses, and real-time data with minimal memory footprint.

Real-World Use

A video transcoding server reads a 4GB video file via Readable stream, transforms chunks through a codec Transform, and writes to S3 via a Writable stream, using under 50MB memory.

Streams Deep Path

flowchart LR
  A[Streams Basics] --> B[Streams Deep]
  B --> C[Backpressure]
  B --> D[Pipeline]
  C --> E[Error Handling]
  B --> F{You Are Here}
  style F fill:#f90,color:#fff

Custom Readable Stream

Implement _read() to push data on demand. The stream pulls data when the consumer requests it.

const { Readable } = require("node:stream");
class CounterStream extends Readable {
  constructor(max = 10) {
    super({ objectMode: true });
    this.max = max;
    this.current = 1;
  }
  _read() {
    if (this.current <= this.max) {
      this.push({ count: this.current++ });
    } else {
      this.push(null);
    }
  }
}
const counter = new CounterStream(5);
counter.on("data", (item) => console.log("Received:", item));

Custom Writable Stream

Implement _write() to Process each chunk. The stream signals readiness via the callback.

const { Writable } = require("node:stream");
const fs = require("node:fs");
class FileWriteStream extends Writable {
  constructor(filePath) {
    super({ highWaterMark: 16384 });
    this.fd = fs.openSync(filePath, "w");
  }
  _write(chunk, encoding, callback) {
    fs.write(this.fd, chunk, 0, chunk.length, null, (err) => {
      callback(err);
    });
  }
  _final(callback) {
    fs.close(this.fd, callback);
  }
}
const ws = new FileWriteStream("/tmp/output.txt");
ws.write("Hello ");
ws.write("World ");
ws.end();

Custom Transform Stream

Implement _transform() to modify each chunk. Also implement _flush() for remaining data.

const { Transform } = require("node:stream");
class UpperCaseTransform extends Transform {
  _transform(chunk, encoding, callback) {
    this.push(chunk.toString().toUpperCase());
    callback();
  }
  _flush(callback) {
    this.push("---END---");
    callback();
  }
}
const upper = new UpperCaseTransform();
upper.on("data", (chunk) => console.log(chunk.toString()));
upper.write("hello");
upper.write("world");
upper.end();

Object Mode Streams

Streams can work with JavaScript objects instead of Buffers by enabling objectMode.

const { Transform } = require("node:stream");
const batchProcessor = new Transform({
  objectMode: true,
  transform(chunk, encoding, callback) {
    callback(null, { ...chunk, processed: true, timestamp: Date.now() });
  },
});
batchProcessor.on("data", (obj) => console.log("Processed:", obj));
batchProcessor.write({ id: 1, name: "Alice" });
batchProcessor.write({ id: 2, name: "Bob" });
batchProcessor.end();

Custom Duplex Stream

Duplex implements both Readable and Writable. Useful for network sockets and protocol handlers.

const { Duplex } = require("node:stream");
class EchoDuplex extends Duplex {
  constructor() {
    super({ allowHalfOpen: false });
    this.buffer = [];
  }
  _write(chunk, encoding, callback) {
    this.buffer.push(chunk);
    callback();
  }
  _read(size) {
    while (this.buffer.length) {
      if (!this.push(this.buffer.shift())) break;
    }
    if (this._writableState.finished && !this.buffer.length) {
      this.push(null);
    }
  }
}

Common Mistakes

1. Not Handling Backpressure

Writing faster than the consumer can process causes unbounded memory growth. Respect the return value of write().

2. Using Streams When Not Needed

Small files under 50MB are fine with readFile/writeFile. Streams add complexity without benefit for small data.

3. Forgetting Error Handling in Custom Streams

Custom _write and _transform must call callback with errors. Unhandled errors leave the stream in an unknown state.

4. Confusing Object Mode and Buffer Mode

objectMode streams do not handle binary data correctly. Use Buffer mode for binary and objectMode for JS objects.

5. Not Implementing _final in Writable Streams

The _final callback handles cleanup when the stream ends. Without it, resources like file handles leak.

Practice Questions

1. What are the four stream types in Node.js?

Readable (data source), Writable (data sink), Transform (modify data), Duplex (both read and write).

2. What is the purpose of the highWaterMark option?

It controls the internal buffer size (default 16KB for binary, 16 objects for objectMode), triggering backpressure when exceeded.

3. When should you use a Transform instead of a Duplex?

Use Transform when input and output are logically related (modified version of input). Use Duplex when input and output are independent.

4. What does _flush do in a Transform stream?

Called before the stream ends. Push remaining data or final transformations in _flush.

5. Challenge: Create a LineReader Transform that emits full lines from a stream.

class LineReader extends Transform {
  constructor() { super({ objectMode: 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();
  }
  _flush(cb) { if (this.buffer) this.push(this.buffer); cb(); }
}

FAQ

Mini Project: File Compression Pipeline

Build a pipeline that reads a file, compresses with gzip, and writes to a new file.

const { pipeline } = require("node:stream/promises");
const fs = require("node:fs");
const zlib = require("node:zlib");
async function compressFile(inputPath, outputPath) {
  try {
    await pipeline(
      fs.createReadStream(inputPath),
      zlib.createGzip(),
      fs.createWriteStream(outputPath)
    );
    console.log("File compressed successfully");
  } catch (err) {
    console.error("Pipeline failed:", err.message);
  }
}
compressFile(__filename, __filename + ".gz");

What's Next

Node.js Stream Backpressure Node.js Stream Pipeline Node.js Error Handling