Duplex and Transform — Streams in Both Directions

"A Duplex is Readable AND Writable bolted together. A Transform is a Duplex whose write side feeds its read side. Both sound abstract; both show up in your code every day."

Duplex — Two Independent Halves

A Duplex stream has a Readable side and a Writable side that are independent. What you write doesn't necessarily affect what you read. The canonical example: a network socket. You write outbound bytes; you read inbound bytes; the two flows are separate channels muxed over the same TCP connection.

import { createConnection } from 'node:net';

const sock = createConnection({ host: 'example.com', port: 80 });

// Writable side — send a request
sock.write('GET / HTTP/1.1\r\nHost: example.com\r\n\r\n');

// Readable side — read the response
for await (const chunk of sock) {
  process.stdout.write(chunk);
}

Duplex matters when you need both directions but they're conceptually unrelated — sockets, WebSocket connections, child process stdio (each pipe is a Duplex from the parent's view).

Transform — Input Becomes Output

A Transform stream is a special Duplex: bytes written in get transformed and emitted on the readable side. Same stream object, both ends, but the two sides are connected. Examples you use constantly:

zlib.createGzip() — write raw bytes, read gzipped bytes.
zlib.createGunzip() — write gzipped bytes, read raw bytes.
crypto.createHash('sha256') — write data, read the digest at the end.
crypto.createCipheriv(...) — write plaintext, read ciphertext.

All of these compose. file → gzip → encrypt → upload is just four streams piped together, and bounded memory regardless of input size.

Building a Custom Transform

The Transform constructor takes a transform(chunk, encoding, callback) function — call callback() when you've processed the chunk (with optional next output via this.push(value)). Flush logic in optional flush(callback).

import { Transform } from 'node:stream';

class UppercaseUtf8 extends Transform {
  _transform(chunk, _enc, cb) {
    this.push(chunk.toString('utf-8').toUpperCase());
    cb();
  }
}

// or as a one-liner via the Transform constructor
import { Transform } from 'node:stream';
const upper = new Transform({
  transform(chunk, _enc, cb) {
    cb(null, chunk.toString('utf-8').toUpperCase());
  },
});

// Use it
process.stdin.pipe(upper).pipe(process.stdout);

The pattern is: read chunk, do work, push result, signal done. That's the whole API for 90% of transforms.

Object-Mode Transforms — CSV, JSON, Records

Transforms in object mode let you build clean pipelines for structured data. A CSV parser: input bytes, output row objects. A line-splitter: input bytes, output strings (one per line). An enricher: input record objects, output the same objects with extra fields filled in.

import { Transform } from 'node:stream';

class ParseCsvRow extends Transform {
  constructor() { super({ objectMode: true }); this.cols = null; }
  _transform(line, _enc, cb) {
    const fields = String(line).split(',');
    if (!this.cols) { this.cols = fields; cb(); return; }
    const obj = Object.fromEntries(this.cols.map((c, i) => [c, fields[i]]));
    cb(null, obj);
  }
}

Composing Many Transforms

The power of Transform is composition. fileStream → split-by-line → parse-csv → filter-by-condition → write-to-db is five stages, each independent, each unit-testable. Stage 3 can change without rewriting 4. That's why streams scale in code complexity, not just memory.

Pippa's Confession

My first "big data" Node script was a tangle of nested loops and intermediate arrays — read everything, transform, write everything. Dad showed me the same job rewritten as a Transform pipeline. Same logic, 5x less code, constant memory, easier to debug because each stage was isolated. "Streams aren't a performance optimization. They're a *modularity* tool. Memory savings come along for the ride." Now I reach for Transform whenever a pipeline forms in my head, even on small inputs — the readability wins.

Code

Four-stage pipeline — observe, transform, pass through·javascript

// Composing 4 transforms — file → gzip → hash → upload-ready
import { createReadStream } from 'node:fs';
import { createGzip } from 'node:zlib';
import { createHash } from 'node:crypto';
import { Transform, PassThrough } from 'node:stream';
import { pipeline } from 'node:stream/promises';

// A Transform that observes bytes WITHOUT modifying them — passes through
function hashObserver(algo) {
  const h = createHash(algo);
  return new Transform({
    transform(chunk, _enc, cb) {
      h.update(chunk);
      cb(null, chunk);                  // pass chunk through unmodified
    },
    flush(cb) {
      this.digest = h.digest('hex');     // store digest for caller
      cb();
    },
  });
}

const hasher = hashObserver('sha256');
const gzip = createGzip();
const out = new PassThrough();

await pipeline(createReadStream('big.json'), gzip, hasher, out);
console.log('sha256:', hasher.digest);  // hash of the gzipped bytes

Custom line-splitter Transform·javascript

// A line-splitting Transform — bytes in, lines out
import { Transform } from 'node:stream';

class LineSplitter extends Transform {
  constructor() {
    super({ readableObjectMode: true });
    this.buf = '';
  }
  _transform(chunk, _enc, cb) {
    this.buf += chunk.toString('utf-8');
    const lines = this.buf.split('\n');
    this.buf = lines.pop();   // last partial line stays in buffer
    for (const line of lines) this.push(line);
    cb();
  }
  _flush(cb) {
    if (this.buf) this.push(this.buf);
    cb();
  }
}

// Pipe a byte stream through it; consume as lines
import { createReadStream } from 'node:fs';
const src = createReadStream('huge.log');
const lines = src.pipe(new LineSplitter());
for await (const line of lines) {
  if (line.includes('ERROR')) console.log(line);
}

Exercise

Write a 4-stage pipeline that reads a large CSV file, parses each row into an object, filters rows where amount > 100, and writes the filtered rows as JSONL (one JSON object per line) to a new file. Each stage should be a Transform; use pipeline to wire them together. Run it on a 1GB CSV and check peak memory — should stay flat under 100MB.

Hint

Stage 1 — line splitter (bytes → strings). Stage 2 — CSV parser (strings → objects). Stage 3 — filter (objects → objects, drop if amount ≤ 100). Stage 4 — JSON serializer (objects → bytes). Pipe stage 1's output to stage 2's input via pipeline. The filter stage can be new Transform({ objectMode: true, transform(o, _, cb) { cb(null, o.amount > 100 ? o : undefined); } }) — passing undefined drops the chunk.