Streams and Pipelines

Stream composition for large IO - prefer pipeline over manual pipe for error propagation. Results appear in the same fence: same-line // comments when short, multiline // blocks below the sample when not.

Readable from Iterable

Build a readable from an async or sync iterable for testing and generators.

import { Readable } from "node:stream";
const r = Readable.from(["a", "b", "c"]);
const chunks: string[] = [];
for await (const chunk of r) chunks.push(String(chunk));
chunks  // ["a", "b", "c"]

pipeline Promise

stream/promises.pipeline destroys streams on error and returns a promise you can await.

import { pipeline } from "node:stream/promises";
import { Readable, Writable } from "node:stream";
const out: string[] = [];
await pipeline(
  Readable.from(["hi"]),
  new Writable({ write(c, _e, cb) { out.push(String(c)); cb(); } }),
);
out  // ["hi"]

Transform Map Chunks

Transform streams map or filter chunks in the middle of a pipeline.

import { Transform, Readable } from "node:stream";
const upper = new Transform({
  transform(chunk, _enc, cb) {
    cb(null, String(chunk).toUpperCase());
  },
});
Readable.from(["ab"]).pipe(upper);
// upper emits "AB"

Backpressure write

Honor write returning false and wait for drain before writing more.

function write(stream: NodeJS.WritableStream, data: string) {
  if (!stream.write(data)) {
    return new Promise<void>((res) => stream.once("drain", () => res()));
  }
}
// returns Promise when buffer is full, else undefined

Finished Helper

Wait until a stream is no longer readable/writable with finished.

import { finished } from "node:stream/promises";
import { Readable } from "node:stream";
const rs = Readable.from(["x"]);
rs.resume();
await finished(rs);
// resolves when stream ends (no throw)

objectMode Stream

Object mode passes JS values instead of buffers/strings - great for internal pipelines.

import { Readable } from "node:stream";
const r = Readable.from([{ id: 1 }, { id: 2 }], { objectMode: true });
const first = await r[Symbol.asyncIterator]().next();
first.value  // { id: 1 }

PassThrough Tee

PassThrough is a no-op transform useful for tapping or testing.

import { PassThrough } from "node:stream";
const tap = new PassThrough();
let n = 0;
tap.on("data", (c) => { n += c.length; });
tap.end("ab");
// after end: n === 2

Readable async iterator

Consume any readable with for await - remember it fully reads the stream.

import { Readable } from "node:stream";
const parts: string[] = [];
for await (const chunk of Readable.from(["1", "2"])) {
  parts.push(String(chunk));
}
parts  // ["1", "2"]

Writable writev

Custom writables can implement _writev to batch chunks efficiently.

// In a custom Writable subclass:
// _writev(chunks, cb) { /* flush many chunks */ cb(); }
// Node calls _writev when multiple chunks are queued

Destroy on Error

Always destroy streams on failure to free file descriptors and sockets.

import { Readable } from "node:stream";
const rs = Readable.from([]);
rs.on("error", (err) => {
  rs.destroy();
  err instanceof Error  // true
});

zlib gzip gunzip

Compression streams compose cleanly in pipelines.

import { pipeline } from "node:stream/promises";
import { Readable, PassThrough } from "node:stream";
import { createGzip, createGunzip } from "node:zlib";
const sink = new PassThrough();
const chunks: Buffer[] = [];
sink.on("data", (c) => chunks.push(c));
await pipeline(Readable.from(["hello"]), createGzip(), createGunzip(), sink);
Buffer.concat(chunks).toString()  // "hello"

High Water Mark

Tune buffering with highWaterMark when default 16KiB is wrong for your workload.

import { Readable } from "node:stream";
const rs = Readable.from(["x"], { highWaterMark: 1 });
rs.readableHighWaterMark  // 1

Duplex Net Socket

Sockets are duplex streams - readable and writable ends of one connection.

// const socket = net.connect(port);
// socket.write("ping\n");
// for await (const chunk of socket) handle(chunk);
// socket is both Readable and Writable

pipeline with AbortSignal

Cancel long pipelines with AbortSignal on modern Node.

import { pipeline } from "node:stream/promises";
import { Readable, PassThrough } from "node:stream";
const ac = new AbortController();
ac.abort();
try {
  await pipeline(Readable.from(["a"]), new PassThrough(), { signal: ac.signal });
} catch (e) {
  (e as Error).name  // "AbortError"
}

Compose Utility

stream.compose builds a duplex from a sequence of streams (Node 16+ patterns).

import { compose, PassThrough } from "node:stream";
const duplex = compose(new PassThrough(), new PassThrough());
typeof duplex.pipe  // "function"

Buffers and Encoding - chunk types flowing through streams
Filesystem and Path - createReadStream and write streams
HTTP and Fetch - streaming request and response bodies

Node.js API Documentation - Complete reference for core modules and APIs.
Node.js Learn Guides - Official learning path.
Node.js Best Practices (GitHub) - Curated production best practices.
Node.js Design Patterns, 4th Edition - Production-grade application patterns.
Node.js Cookbook, 5th Edition - Practical server-side recipes.

Stack versions: Node.js 24.18.0 (LTS line 24) · TypeScript 5.6+ · npm 10+

Streams and Pipelines

Readable from Iterable

Build a readable from an async or sync iterable for testing and generators.

import { Readable } from "node:stream";
const r = Readable.from(["a", "b", "c"]);
const chunks: string[] = [];
for await (const chunk of r) chunks.push(String(chunk));
chunks  // ["a", "b", "c"]

pipeline Promise

stream/promises.pipeline destroys streams on error and returns a promise you can await.

import { pipeline } from "node:stream/promises";
import { Readable, Writable } from "node:stream";
const out: string[] = [];
await pipeline(
  Readable.from(["hi"]),
  new Writable({ write(c, _e, cb) { out.push(String(c)); cb(); } }),
);
out  // ["hi"]

Transform Map Chunks

Transform streams map or filter chunks in the middle of a pipeline.

import { Transform, Readable } from "node:stream";
const upper = new Transform({
  transform(chunk, _enc, cb) {
    cb(null, String(chunk).toUpperCase());
  },
});
Readable.from(["ab"]).pipe(upper);
// upper emits "AB"

Backpressure write

Honor write returning false and wait for drain before writing more.

function write(stream: NodeJS.WritableStream, data: string) {
  if (!stream.write(data)) {
    return new Promise<void>((res) => stream.once("drain", () => res()));
  }
}
// returns Promise when buffer is full, else undefined

Finished Helper

Wait until a stream is no longer readable/writable with finished.

import { finished } from "node:stream/promises";
import { Readable } from "node:stream";
const rs = Readable.from(["x"]);
rs.resume();
await finished(rs);
// resolves when stream ends (no throw)

objectMode Stream

Object mode passes JS values instead of buffers/strings - great for internal pipelines.

import { Readable } from "node:stream";
const r = Readable.from([{ id: 1 }, { id: 2 }], { objectMode: true });
const first = await r[Symbol.asyncIterator]().next();
first.value  // { id: 1 }

PassThrough Tee

PassThrough is a no-op transform useful for tapping or testing.

import { PassThrough } from "node:stream";
const tap = new PassThrough();
let n = 0;
tap.on("data", (c) => { n += c.length; });
tap.end("ab");
// after end: n === 2

Readable async iterator

Consume any readable with for await - remember it fully reads the stream.

import { Readable } from "node:stream";
const parts: string[] = [];
for await (const chunk of Readable.from(["1", "2"])) {
  parts.push(String(chunk));
}
parts  // ["1", "2"]

Writable writev

Custom writables can implement _writev to batch chunks efficiently.

// In a custom Writable subclass:
// _writev(chunks, cb) { /* flush many chunks */ cb(); }
// Node calls _writev when multiple chunks are queued

Destroy on Error

Always destroy streams on failure to free file descriptors and sockets.

import { Readable } from "node:stream";
const rs = Readable.from([]);
rs.on("error", (err) => {
  rs.destroy();
  err instanceof Error  // true
});

zlib gzip gunzip

Compression streams compose cleanly in pipelines.

import { pipeline } from "node:stream/promises";
import { Readable, PassThrough } from "node:stream";
import { createGzip, createGunzip } from "node:zlib";
const sink = new PassThrough();
const chunks: Buffer[] = [];
sink.on("data", (c) => chunks.push(c));
await pipeline(Readable.from(["hello"]), createGzip(), createGunzip(), sink);
Buffer.concat(chunks).toString()  // "hello"

High Water Mark

Tune buffering with highWaterMark when default 16KiB is wrong for your workload.

import { Readable } from "node:stream";
const rs = Readable.from(["x"], { highWaterMark: 1 });
rs.readableHighWaterMark  // 1

Duplex Net Socket

Sockets are duplex streams - readable and writable ends of one connection.

// const socket = net.connect(port);
// socket.write("ping\n");
// for await (const chunk of socket) handle(chunk);
// socket is both Readable and Writable

pipeline with AbortSignal

Cancel long pipelines with AbortSignal on modern Node.

import { pipeline } from "node:stream/promises";
import { Readable, PassThrough } from "node:stream";
const ac = new AbortController();
ac.abort();
try {
  await pipeline(Readable.from(["a"]), new PassThrough(), { signal: ac.signal });
} catch (e) {
  (e as Error).name  // "AbortError"
}

Compose Utility

stream.compose builds a duplex from a sequence of streams (Node 16+ patterns).

import { compose, PassThrough } from "node:stream";
const duplex = compose(new PassThrough(), new PassThrough());
typeof duplex.pipe  // "function"

Buffers and Encoding - chunk types flowing through streams
Filesystem and Path - createReadStream and write streams
HTTP and Fetch - streaming request and response bodies

Node.js API Documentation - Complete reference for core modules and APIs.
Node.js Learn Guides - Official learning path.
Node.js Best Practices (GitHub) - Curated production best practices.
Node.js Design Patterns, 4th Edition - Production-grade application patterns.
Node.js Cookbook, 5th Edition - Practical server-side recipes.

Stack versions: Node.js 24.18.0 (LTS line 24) · TypeScript 5.6+ · npm 10+

Streams and Pipelines

Search Documentation

Streams and Pipelines