Duplex & Transform Streams

The two remaining stream types combine reading and writing. A Duplex stream is readable and writable with two independent channels — a TCP socket is the classic example. A Transform is a special duplex where what you write comes back out modified — gzip, encryption, and parsers are all transforms. These are the building blocks of stream pipelines.

Duplex: two independent channels

A duplex's readable and writable sides are separate. Writing to it does not feed its readable side — they're like two pipes bundled together. A network socket is the perfect mental model: what you write() goes out to the peer; what you read in is what the peer sent — totally unrelated streams of bytes.

import { connect } from 'node:net'

const socket = connect(80, 'example.com')   // a Duplex stream

// Writable side — send to the server:
socket.write('GET / HTTP/1.0\r\n\r\n')

// Readable side — receive from the server (independent):
for await (const chunk of socket) {
  process.stdout.write(chunk)
}

Duplex = readable + writable, NOT connected to each other

The defining trait: the input and output sides are independent. Contrast this with a Transform, where the output *is* a function of the input. If you find yourself wanting "what I write should come back changed," you want a Transform, not a raw Duplex.

Transform: write in, modified data out

A Transform stream sits in the middle of a pipeline: data flows in the writable side, your transform function processes it, and the result flows out the readable side. This is the most common custom stream you'll write.

import { Transform } from 'node:stream'

const upperCase = new Transform({
  transform(chunk, _encoding, callback) {
    // push the modified chunk downstream...
    this.push(chunk.toString().toUpperCase())
    callback()                 // ...then signal "ready for next chunk"
  },
})

process.stdin.pipe(upperCase).pipe(process.stdout)

Call `callback()` exactly once per chunk — and use it to emit too

Like a writable's `_write`, a transform's `transform(chunk, enc, callback)` must call `callback` once when the chunk is processed — omit it and the pipeline stalls. You emit output either via `this.push(value)` (any number of times, including zero) or by passing data as the second arg: `callback(null, output)`. Calling `callback(err)` propagates an error down the pipeline.

transform() vs flush()

transform runs per chunk; the optional flush runs once after the last chunk, before the stream ends — the place to emit anything you've been buffering (a final delimiter, a checksum, a remaining partial record):

import { Transform } from 'node:stream'

// Buffer partial lines across chunks, emit complete lines:
const splitLines = new Transform({
  readableObjectMode: true,
  transform(chunk, _enc, cb) {
    this._buf = (this._buf || '') + chunk.toString()
    const parts = this._buf.split('\n')
    this._buf = parts.pop()                // keep the trailing partial line
    for (const line of parts) this.push(line)
    cb()
  },
  flush(cb) {
    if (this._buf) this.push(this._buf)    // emit the final partial line
    cb()
  },
})

`flush` is for trailing state

Because chunk boundaries are arbitrary, a transform often holds leftover bytes between calls (here, an unterminated line). Whatever remains when input ends would be lost without `flush` — it's your last chance to emit. Anytime your transform buffers across chunks, you almost certainly need a `flush`.

Asymmetric object modes

A transform can have different modes on each side — bytes in, objects out, or vice versa. This is how parsers work: consume a byte stream, emit structured records.

import { Transform } from 'node:stream'

// Bytes in (writable side), objects out (readable side):
const parseJsonLines = new Transform({
  writableObjectMode: false,   // accept Buffer/string chunks
  readableObjectMode: true,    // emit parsed objects
  transform(chunk, _enc, cb) {
    for (const line of chunk.toString().split('\n')) {
      if (line.trim()) this.push(JSON.parse(line))
    }
    cb()
  },
})

Option	Effect
`objectMode: true`	Both sides handle objects
`writableObjectMode`	Input side accepts objects
`readableObjectMode`	Output side emits objects

Built-in transforms you already use

You rarely need a custom transform for common jobs — Node ships fast, battle-tested ones. They slot straight into a pipeline:

import { createReadStream, createWriteStream } from 'node:fs'
import { createGzip } from 'node:zlib'
import { pipeline } from 'node:stream/promises'

// zlib.createGzip() is a Transform: bytes in → compressed bytes out
await pipeline(
  createReadStream('app.log'),
  createGzip(),                      // the transform in the middle
  createWriteStream('app.log.gz'),
)

Module	Transforms
`zlib`	`createGzip` / `createGunzip` / `createDeflate` / `createBrotliCompress`
`crypto`	`createCipheriv` / `createDecipheriv` / `createHash` (hash is writable→readable)
`node:stream`	`Transform`, `PassThrough` (a no-op transform)

`PassThrough` — the no-op transform

`stream.PassThrough` reads input and emits it unchanged. It sounds useless but is handy for *observing* a stream (count bytes, tee data) without altering it, or as a placeholder/adapter when an API hands you a stream and you need a writable to attach to. Think of it as a transform whose transform function is the identity.

The simplest way: a generator transform

Modern Node lets you drop an async generator directly into pipeline as a transform stage — no Transform subclass, no callbacks, just yield:

import { pipeline } from 'node:stream/promises'
import { createReadStream, createWriteStream } from 'node:fs'

await pipeline(
  createReadStream('in.txt', 'utf8'),
  async function* (source) {              // a transform, as a generator
    for await (const chunk of source) {
      yield chunk.toUpperCase()           // each yield is an output chunk
    }
  },
  createWriteStream('out.txt'),
)

Prefer generator stages for new code

An async generator stage gets backpressure, error propagation, and cleanup from `pipeline` for free, with far less boilerplate than a `Transform` subclass. Reach for a full `Transform` class only when you need `flush` semantics, custom `highWaterMark`, or to expose a reusable stream object.

Quick reference

Need	Use
Independent read + write (socket-like)	`Duplex`
Modify data in flight	`Transform` or generator stage
Emit trailing/buffered data	`flush(cb)`
Parse bytes → records	Asymmetric object mode
Observe without changing	`PassThrough`
Compress / encrypt	`zlib` / `crypto` built-ins

Connect these stages together the classic way — and learn why it's tricky: [Piping Streams](/nodejs/piping-streams).