Streams

A pull/push source with backpressure: enqueue via a controller in start/pull, consume with for await, .getReader().read(), or by piping. ReadableStream.from() builds one from an array. Runs on the promise/microtask runtime alone — no fibers or libcurl for in-memory pipelines.

ReadableStream<T> — A readable stream.

let n = 0
const numbers = new ReadableStream<number>({
  pull: (controller) => {
    n = n + 1
    if (n > 5) { controller.close() } else { controller.enqueue(n * n) }
  }
}, { highWaterMark: 1 })
for await (const square of numbers) { console.log("square:", square) }

A backpressured sink: .write() queues a chunk (each write carries its own promise), .close()/.abort() finish it, and ready re-arms on backpressure flips. The sink's write/close/abort callbacks drive delivery.

WritableStream<T> — A writable stream.

const printed: string[] = []
const sink = new WritableStream<string>({
  write: (chunk) => { printed.push(chunk) }
})
await source.pipeThrough(upper).pipeTo(sink)

A writable-in / readable-out pair whose transform callback maps chunks from one side to the other, running when the readable side has capacity (else parking one chunk). Compose it with pipeThrough. Writable HWM defaults to 1, readable HWM to 0.

TransformStream<I, O> — A transform stream.

const upper = new TransformStream<string, string>({
  transform: (chunk, controller) => { controller.enqueue(chunk.toUpperCase()) }
})
await source.pipeThrough(upper).pipeTo(sink)

TransformStreams backed by native zlib: gzip/deflate/deflate-raw compress and decompress a Uint8Array chunk stream. Inflate auto-detects gzip/zlib wrappers, and -lz is linked only when one of these is actually constructed.

TransformStream — A zlib-backed transform stream.

const [rawBranch, decodeBranch] = source.pipeThrough(new CompressionStream("gzip")).tee()
for await (const chunk of decodeBranch.pipeThrough(new DecompressionStream("gzip"))) {
  out = out + dec.decode(chunk)
}

Returns a ReadableStream<Uint8Array> over the blob's bytes — one owned-copy chunk then closed (an empty blob yields no chunk). Consumable via for await or .getReader().

ReadableStream<Uint8Array> — A readable stream of the blob's bytes.

Node's stream classes, ported from Node v24's own implementation to TypeScript and compiled with the program: the same states, event order and process.nextTick deferrals. The options form binds this to the stream, subclasses override _read/_write/_transform/_final/_destroy, chunks are Buffers (or strings with setEncoding, or any value in object mode), and Readable.from/.toWeb/.fromWeb bridge to iterables, async iterables and web streams. The readable operators (map with concurrency, filter, flatMap, drop, take, forEach, toArray, some, every, find, reduce), the signal option and addAbortSignal, Duplex.from and compose are Node's own.

Readable | Writable | Transform | PassThrough — A Node stream.

class Greeter extends Readable {
  words: string[] = ["kalimera", "kosme", "apo", "thessaloniki"]
  i: number = 0
  constructor() { super({ objectMode: true }) }
  _read() {
    if (this.i >= this.words.length) { this.push(null) }
    else { this.push(this.words[this.i]); this.i = this.i + 1 }
  }
}

The callback forms: pipeline(a, ...b, cb) pipes each stream into the next and calls cb once the last has finished, or with the first error, destroying every stream; a stage may be an async generator function, the source any iterable, and the last stage may return a promise, with { signal, end } options; finished(s, [options,] cb) calls cb when the stream has ended, finished and closed, and returns a function removing its listeners. Their promise twins live in stream/promises.

Stream | () => void — pipeline returns the last stream; finished returns its cleanup function.

finished(echo, (err) => { console.log("echo finished, clean:", !err) })
pipeline(src, new PassThrough(), sink, (err) => { console.log("pipeline done", err ? err.message : "ok") })

The promise-returning pipeline and finished, over the callback forms: pipeline resolves once the last stream has finished, finished once the stream has ended and closed; either rejects with the stream's error.

Promise<void> — Resolves on completion, rejects on error.

import { pipeline, finished } from 'stream/promises'
await pipeline(numbers, square, sink)
await finished(words)

Two Duplex instances wired to each other: what one side writes, the other reads. A write completes when the peer reads it, and end() finishes once the peer has ended, as Node's DuplexSide does. Returned as a two-element array.

[Duplex, Duplex] — A pair of cross-wired Duplex streams.

const [clientSide, serverSide] = duplexPair()
serverSide.on("data", (d) => { console.log("server saw:", d.toString()) })
clientSide.end("hello over the pair")