streams — Streams

Processing data incrementally in Node.js — Readable, Writable, Transform streams, pipeline, and async iteration.

What it is, in plain English: Streams are Node.js's way of processing data piece by piece, without loading the whole thing into memory. A Readable stream produces data. A Writable stream consumes data. A Transform stream does both — it receives data, transforms it, and outputs the result. pipeline() connects streams together and handles errors and cleanup. Streams implement the async iterable protocol — you can use for await...of.

Why streams — memory-efficient data processing

Without streams, to process a large file you load it all into memory. A 10GB log file would need 10GB of RAM. Streams process data in small chunks — the same task uses only kilobytes of memory.

The four stream types

// Readable  — source of data
// Writable  — destination for data
// Transform — reads, transforms, writes (both ends)
// Duplex    — separate read and write (like a TCP socket)

// You use streams constantly without realising it:
process.stdin;                      // Readable
process.stdout;                     // Writable
fs.createReadStream("file.txt");    // Readable
fs.createWriteStream("out.txt");    // Writable
zlib.createGzip();                  // Transform
http.IncomingMessage;               // Readable (req in createServer)
http.ServerResponse;                // Writable (res in createServer)

Reading a stream with async iteration

import { createReadStream } from "fs";
import { createInterface }  from "readline";

// Read a large file line by line (memory-efficient):
const rl = createInterface({
  input:     createReadStream("large.csv"),
  crlfDelay: Infinity,
});

let lineNumber = 0;
for await (const line of rl) {
  lineNumber++;
  if (lineNumber === 1) continue;  // skip header
  processRow(line.split(","));
}
console.log(`Processed ${lineNumber} lines`);

pipeline — the correct way to connect streams

import { createReadStream, createWriteStream } from "fs";
import { createGzip } from "zlib";
import { pipeline }   from "stream/promises";

// Compress a file:
await pipeline(
  createReadStream("big-file.json"),   // read
  createGzip(),                         // compress
  createWriteStream("big-file.json.gz") // write
);
// pipeline: handles backpressure, errors, and cleanup automatically

// ✗ Don't use .pipe() directly — it doesn't handle errors:
// readStream.pipe(gzip).pipe(writeStream);  // if gzip fails, streams don't close!

Try It Yourself

// Streams — process data without loading everything into memory
// Demonstrating the concepts (real streams need Node.js)

// The backpressure problem streams solve:
// Imagine reading a 10GB log file and filtering errors:

// ✗ Wrong way — loads entire 10GB into memory:
// const content = await readFile("huge.log", "utf-8");
// const errors  = content.split("\n").filter(l => l.includes("ERROR"));
// writeFile("errors.log", errors.join("\n"));

// ✓ Right way — streams process one chunk at a time:
// import { createReadStream, createWriteStream } from "fs";
// import { pipeline } from "stream/promises";
// import { Transform } from "stream";
//
// const filterErrors = new Transform({
//   transform(chunk, encoding, callback) {
//     const lines = chunk.toString().split("\n")
//       .filter(line => line.includes("ERROR"));
//     if (lines.length) this.push(lines.join("\n") + "\n");
//     callback();
//   }
// });
//
// await pipeline(
//   createReadStream("huge.log"),
//   filterErrors,
//   createWriteStream("errors.log")
// );

// Stream interfaces — what each type does:
const STREAM_TYPES = {
  "Readable":  "Produces data: fs.createReadStream, http.IncomingMessage, process.stdin",
  "Writable":  "Consumes data: fs.createWriteStream, http.ServerResponse, process.stdout",
  "Transform": "Both: zlib.createGzip, crypto.createCipher, custom data processing",
  "Duplex":    "Both read and write, independent: net.Socket, crypto streams",
  "PassThrough": "Transform that passes data unchanged: useful for monitoring",
};

Object.entries(STREAM_TYPES).forEach(([type, desc]) => {
  console.log(\`\n\${type}:\n  \${desc}\`);
});

// Async iteration (cleanest API for reading streams):
// for await (const chunk of readableStream) {
//   process(chunk);
// }

// Creating a simple Readable:
import { Readable } from "stream";
const readable = Readable.from(["chunk1\n", "chunk2\n", "chunk3\n"]);
for await (const chunk of readable) {
  process.stdout.write(chunk);
}

Creating Transform streams and backpressure

Custom Transform stream

import { Transform } from "stream";

// Filter lines containing a pattern:
class LineFilter extends Transform {
  constructor(pattern) {
    super();
    this.pattern  = pattern;
    this.leftover = "";  // handle chunks that split mid-line
  }

  _transform(chunk, encoding, callback) {
    const lines = (this.leftover + chunk.toString()).split("\n");
    this.leftover = lines.pop();  // last (possibly incomplete) line

    for (const line of lines) {
      if (line.includes(this.pattern)) {
        this.push(line + "\n");  // push to output
      }
    }
    callback();  // signal we're done with this chunk
  }

  _flush(callback) {
    if (this.leftover && this.leftover.includes(this.pattern)) {
      this.push(this.leftover + "\n");
    }
    callback();
  }
}

// Use it:
await pipeline(
  createReadStream("server.log"),
  new LineFilter("ERROR"),
  createWriteStream("errors.log")
);

Object mode streams

// Default streams work with Buffers/strings.
// Object mode lets you stream any JS values:
import { Transform } from "stream";
import { createReadStream } from "fs";

class CSVParser extends Transform {
  constructor() {
    super({ objectMode: true });  // ← output objects, not strings
    this.headers = null;
  }

  _transform(chunk, enc, cb) {
    const lines = chunk.toString().split("\n").filter(Boolean);
    for (const line of lines) {
      const values = line.split(",");
      if (!this.headers) { this.headers = values; continue; }
      const obj = Object.fromEntries(this.headers.map((h, i) => [h, values[i]]));
      this.push(obj);  // push a JS object
    }
    cb();
  }
}

// Downstream can now work with objects:
const parser = new CSVParser();
parser.on("data", (row) => {
  console.log(row);  // { name: "Alice", age: "30", city: "Mumbai" }
});

Backpressure, highWaterMark, and streams spec

// Backpressure — what happens when the consumer is slower than the producer:
const fastReadable = createReadStream("huge.log", { highWaterMark: 64 * 1024 }); // 64KB chunks
const slowWritable = createWriteStream("out.log");

// .write() returns false when the internal buffer is full:
fastReadable.on("data", (chunk) => {
  const ok = slowWritable.write(chunk);
  if (!ok) {
    fastReadable.pause();   // stop reading until writable drains
    slowWritable.once("drain", () => fastReadable.resume());
  }
});

// pipeline() handles all of this automatically — use it!
// This is why pipeline > manual pipe > manual event handling.

// highWaterMark — internal buffer size (bytes for binary, count for objectMode):
const readable = new Readable({ highWaterMark: 16 * 1024 }); // 16KB buffer
// When buffer > hwm → stream enters "flowing=false" (backpressure applied)

// WHATWG Streams API (browser-compatible, also in Node.js 16+):
// ReadableStream, WritableStream, TransformStream
// response.body is a ReadableStream — can be piped with pipeTo()

Official Sources

The Codex links only to official documentation and the ECMAScript specification. No third-party tutorials.