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.

Where this bites you

Chunk boundaries ignore your data's structure. A file stream hands you 64 KB at a time, so a CSV row is cut in half whenever it straddles two chunks. A Transform that splits each chunk on "\n" on its own produces one wrong row from the first half and one garbage row from the second. Small test files fit in one chunk and never show it.

import { Transform } from "stream";
import { createReadStream } from "fs";

class CSVParser extends Transform {
  constructor() {
    super({ objectMode: true });
    this.headers = null;
  }
  _transform(chunk, enc, cb) {
    const lines = chunk.toString().split("\n").filter(Boolean);   // each chunk on its own
    for (const line of lines) {
      const values = line.split(",");
      if (!this.headers) { this.headers = values; continue; }
      this.push(Object.fromEntries(this.headers.map((h, i) => [h, values[i]])));
    }
    cb();
  }
}

// people.csv: a header + 5,000 rows, every city "Mumbai" (94 KB: two chunks, 64 KB + 30 KB)
let count = 0;
const broken = [];
createReadStream("people.csv").pipe(new CSVParser())
  .on("data", (row) => { count++; if (row.city !== "Mumbai") broken.push(row); })
  .on("end", () => { console.log("rows:", count); console.log("broken:", broken); });
// rows: 5001
// broken: [ { name: 'user3507', age: '27', city: 'Mum' },
//           { name: 'bai', age: undefined, city: undefined } ]

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" }
});

The CSVParser above has exactly this problem. The LineFilter before it shows the cure: keep the unfinished last line in this.leftover, prepend it to the next chunk, and process whatever is left in _flush():

class CSVParser extends Transform {
  constructor() {
    super({ objectMode: true });
    this.headers  = null;
    this.leftover = "";                          // carry a split line into the next chunk
  }
  _transform(chunk, enc, cb) {
    const lines = (this.leftover + chunk.toString()).split("\n");
    this.leftover = lines.pop();                 // the last piece may be incomplete
    for (const line of lines) this.#row(line);
    cb();
  }
  _flush(cb) {
    if (this.leftover) this.#row(this.leftover); // the file may not end with "\n"
    cb();
  }
  #row(line) {
    if (!line) return;
    const values = line.split(",");
    if (!this.headers) { this.headers = values; return; }
    this.push(Object.fromEntries(this.headers.map((h, i) => [h, values[i]])));
  }
}
// rows: 5000
// broken: []

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 the buffer reaches hwm, push() returns false and _read() stops being called until it drains

// 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.