Chapter 04

Buffers & Streams

Bytes in memory and data over time — Buffer, the four stream types, backpressure and pipeline.

10 topics

Parts marked Advanced are extra depth. Skip them on a first read or a quick revision.

#Buffers

JavaScript strings are for text. But almost everything Node touches at the edges — files, TCP sockets, HTTP bodies, images, gzip data, hashes, encryption keys — is raw bytes, and bytes aren't text until you decide how to decode them. A Buffer is Node's type for a fixed-length sequence of bytes. It's a subclass of Uint8Array, so each element is an integer from 0 to 255, and its memory lives outside the normal JavaScript object heap (it shows up as arrayBuffers in process.memoryUsage()).

A good mental model is a row of numbered boxes, each holding one byte. The row's length is fixed when it's created. You can read and write boxes by index, view part of the row without copying it, and decode the whole row as text using some encoding (utf8, base64, hex, …). Encoding turns text into bytes; decoding turns bytes back into text.

buffer-basics.cjs

const buffer = Buffer.from("Hello");
console.log(buffer); // <Buffer 48 65 6c 6c 6f>
console.log(buffer.length, buffer[0], buffer.toString("utf-8")); // 5 72 Hello
console.log(buffer instanceof Uint8Array, Buffer.isBuffer(buffer)); // true true

console.log(Buffer.from([72, 105]).toString()); // Hi
console.log(Buffer.from("48656c6c6f", "hex").toString()); // Hello

What’s happening

Buffer.from("Hello") encodes the string as UTF-8 (the default) into new memory: five bytes, one per letter, because these are all ASCII characters.
console.log shows the bytes in hexadecimal: 48 is 72 in decimal, the code for H; 65 is e; 6c is l; 6f is o.
buffer.length is the byte count (5), and buffer[0] reads one byte as a number (72). toString("utf-8") decodes the bytes back into the string Hello ("utf-8" and "utf8" are the same).
It really is a Uint8Array, so typed-array methods (indexOf, subarray, fill, iteration) work on it, and Web APIs that take a Uint8Array accept it.
Buffer.from also accepts an array of byte values or an encoded string: [72, 105] are the bytes for Hi, and the hex string "48656c6c6f" decodes to the same five bytes as "Hello".

AdvancedInitialisation: alloc vs allocUnsafe

There are three ways to get a fresh buffer of a given size, and the difference is about initialisation:

alloc.cjs

const zeros = Buffer.alloc(8); // zero-filled
console.log(zeros); // <Buffer 00 00 00 00 00 00 00 00>

const unsafe = Buffer.allocUnsafe(8); // NOT initialised — may contain old memory
unsafe.fill(0); // overwrite before you read or send it

const b = Buffer.alloc(4);
b[0] = 255;
b[1] = 256; // too big: stored modulo 256
b[2] = 300;
b[3] = -1;
console.log(b); // <Buffer ff 00 2c ff>

b.writeUInt16BE(0x1234, 0); // two bytes, big-endian, at offset 0
console.log(b, b.readUInt16BE(0), b.readUInt16LE(0)); // <Buffer 12 34 2c ff> 4660 13330

What’s happening

Buffer.alloc(8) asks for 8 bytes and zero-fills them. It's always safe, at the cost of writing those zeros.
Buffer.allocUnsafe(8) skips the zero-filling. The memory may still hold bytes from whatever used it before — possibly another request's data. It's faster for large buffers you're about to overwrite completely (for example, reading a file into it); otherwise fill it or use alloc.
Each element is an unsigned byte, so assignments are reduced modulo 256: 256 → 0, 300 → 44 (0x2c), -1 → 255. No error is thrown — an easy way to corrupt data silently.
Multi-byte numbers use the read…/write… methods. writeUInt16BE(0x1234, 0) stores 12 34 — big-endian, most significant byte first, the order network protocols use. Reading the same two bytes as little-endian gives 0x3412 = 13330 instead of 4660: binary formats always specify byte order, and getting it wrong gives plausible-looking garbage.

AdvancedSlices share memory: subarray

Slicing a buffer doesn't copy it — and that surprises people who expect array behaviour:

views.cjs

const whole = Buffer.from("Hello, world");
const view = whole.subarray(0, 5); // a window onto the same memory
view[0] = 0x4a; // 'J'
console.log(whole.toString(), "|", view.toString()); // Jello, world | Jello

const copy = Buffer.from(whole.subarray(7)); // Buffer.from(buffer) copies
copy[0] = 0x57; // 'W'
console.log(whole.toString(), "|", copy.toString()); // Jello, world | World

console.log(Buffer.concat([Buffer.from("ab"), Buffer.from("cd")]).toString()); // abcd
console.log(Buffer.from("abc").equals(Buffer.from("abc")), Buffer.from("abc") === Buffer.from("abc")); // true false

What’s happening

subarray(0, 5) returns a new Buffer object pointing at bytes 0–4 of the same memory. Writing 0x4a through view changes whole too: Hello becomes Jello.
That's why subarray is cheap — no bytes move — and it's how parsers pick apart a packet without allocating. The cost: a small view keeps the whole underlying memory alive, so holding a 10-byte view of a 10 MB buffer keeps 10 MB in memory.
Buffer.from(existingBuffer) copies the bytes, so changing copy leaves whole alone.
Buffer.concat allocates one new buffer and copies all the pieces into it — the standard way to assemble chunks.
=== compares object identity, so two buffers with the same bytes are not ===. Use .equals() (or Buffer.compare) to compare contents; for secrets such as HMACs, use crypto.timingSafeEqual.
buf.slice() behaves like subarray (it shares memory too), unlike Array.prototype.slice and Uint8Array.prototype.slice, which copy. That mismatch is why the docs steer you to subarray.

AdvancedEdge case: pooled allocUnsafe memory

#Why use Buffers instead of strings

A JavaScript string is a sequence of UTF-16 code units, and .length counts those code units — not bytes, and not the characters a person would count. Disks, sockets and protocols deal in bytes, usually UTF-8 encoded. Most of the time the two numbers differ, and treating one as the other gives wrong Content-Length headers, truncated database fields, and broken binary files.

byte-length.cjs

for (const s of ["A", "é", "€", "😀", "héllo"]) {
  console.log(
    JSON.stringify(s).padEnd(8),
    "length", s.length,
    "| utf8 bytes", Buffer.byteLength(s),
    "| utf16le bytes", Buffer.byteLength(s, "utf16le"),
    "|", Buffer.from(s),
  );
}
// "A"      length 1 | utf8 bytes 1 | utf16le bytes 2 | <Buffer 41>
// "é"      length 1 | utf8 bytes 2 | utf16le bytes 2 | <Buffer c3 a9>
// "€"      length 1 | utf8 bytes 3 | utf16le bytes 2 | <Buffer e2 82 ac>
// "😀"     length 2 | utf8 bytes 4 | utf16le bytes 4 | <Buffer f0 9f 98 80>
// "héllo"  length 5 | utf8 bytes 6 | utf16le bytes 10 | <Buffer 68 c3 a9 6c 6c 6f>

What’s happening

"A" is ASCII: one code unit, one UTF-8 byte (41).
"é" is one code unit but two UTF-8 bytes (c3 a9); "€" is one code unit but three (e2 82 ac). UTF-8 uses 1–4 bytes per character depending on the code point.
"😀" is outside the Basic Multilingual Plane, so UTF-16 needs a surrogate pair: .length is 2 for one visible character, and UTF-8 needs 4 bytes.
"héllo" has .length 5 but is 6 bytes on the wire. If you set Content-Length: str.length, the client would cut off the last byte. Buffer.byteLength(str) gives the real size without allocating a buffer.

AdvancedHex, base64 and base64url conversions

Buffers also convert between the binary-to-text encodings you meet in APIs: hex (two characters per byte), base64 (four characters per three bytes) and base64url (the URL-safe variant used in JWTs — see How a JWT is built):

encodings.cjs

const buf = Buffer.from("Hi Rohit!");
console.log(buf.toString("hex")); // 486920526f68697421
console.log(buf.toString("base64")); // SGkgUm9oaXQh
console.log(Buffer.from("SGkgUm9oaXQh", "base64").toString()); // Hi Rohit!

const tricky = Buffer.from([0xfb, 0xff, 0xbf]);
console.log(tricky.toString("base64"), tricky.toString("base64url")); // +/+/ -_-_

What’s happening

hex writes each byte as two hex digits: 9 bytes → 18 characters.
base64 packs every 3 bytes into 4 characters from a 64-character alphabet: 9 bytes → 12 characters. That's the 33% size overhead of base64.
Decoding with Buffer.from(text, "base64") gets the original bytes back exactly.
Bytes fb ff bf happen to produce + and /, which mean something in URLs. base64url substitutes - and _ and drops the = padding — the only difference between the two.

AdvancedBinary through a string loses bytes

The real danger is pushing binary data through a string. Arbitrary bytes are usually not valid UTF-8, and decoding replaces invalid bytes with U+FFFD, losing them for good:

binary-roundtrip.cjs

const bin = Buffer.from([0xff, 0x00, 0x41]); // e.g. three bytes of a PNG
const str = bin.toString("utf8");
console.log(JSON.stringify(str), str.length); // "�\u0000A" 3

const back = Buffer.from(str, "utf8");
console.log(back, back.length, back.equals(bin)); // <Buffer ef bf bd 00 41> 5 false

What’s happening

0xff can never appear in valid UTF-8, so the decoder replaces it with � (U+FFFD, the replacement character). 0x00 and 0x41 decode fine as \u0000 and A.
Encoding the string back gives ef bf bd (the UTF-8 for U+FFFD), then 00 41. The original 0xff is gone and the data is now 5 bytes instead of 3.
An image, a zip file or an encrypted payload that goes through body += chunk (string concatenation calls toString() on each chunk) is corrupted in exactly this way. Keep binary as Buffers and Buffer.concat them.

AdvancedSplit characters and StringDecoder

Even for real text, decoding chunk by chunk goes wrong when a multi-byte character straddles a chunk boundary — which, with 64 KiB chunks, happens all the time:

chunk-boundary.cjs

const { StringDecoder } = require("node:string_decoder");

const bytes = Buffer.from("€5"); // e2 82 ac 35
const chunk1 = bytes.subarray(0, 2); // e2 82   — first two bytes of the €
const chunk2 = bytes.subarray(2); //    ac 35

console.log("naive:", JSON.stringify(chunk1.toString() + chunk2.toString())); // "��5"

const decoder = new StringDecoder("utf8");
console.log(JSON.stringify(decoder.write(chunk1)), JSON.stringify(decoder.write(chunk2))); // "" "€5"

What’s happening

The € is three bytes, and the chunk boundary falls after the second one.
chunk1.toString() sees an incomplete sequence and emits a replacement character; chunk2.toString() sees a stray continuation byte (ac) and emits another. Result "��5" — corrupted, though every byte arrived.
StringDecoder remembers incomplete bytes between calls. write(chunk1) returns "" and holds back e2 82; write(chunk2) prepends them and returns the complete "€5".
Readable streams use a StringDecoder internally when you call stream.setEncoding("utf8") or pass { encoding: "utf8" }, so chunks arrive as correctly decoded strings. If you get Buffer chunks and need text, either do that or Buffer.concat first and decode once.

AdvancedCorrecting my notes: string memory size

Correcting my notes

My notes said strings take up "two bytes per character", so text is cheaper as a Buffer. That's too strong. JavaScript exposes strings as UTF-16 code units, but V8 stores a string as one byte per character when every character fits in Latin-1 (code points 0–255), and switches to two bytes only when it has to. Measured on Node 26.7 with --expose-gc:

string-memory.cjs

// run with: node --expose-gc string-memory.cjs
const N = 10_000_000;
const used = () => {
  gc();
  return process.memoryUsage().heapUsed;
};
function build(char) {
  let s = "";
  for (let i = 0; i < 1000; i++) s += char.repeat(N / 1000); // a "rope" of 1000 pieces
  return JSON.parse(JSON.stringify(s)); // copy it into one flat string; the rope becomes garbage
}
function measure(label, char) {
  const before = used();
  const s = build(char);
  console.log(label, `~${((used() - before) / s.length).toFixed(2)} bytes per char`);
  return s;
}
const keep = [measure('"a" (ASCII)  ', "a"), measure('"é" (Latin-1)', "é"), measure('"€" (U+20AC) ', "€")];
// "a" (ASCII)   ~1.00 bytes per char
// "é" (Latin-1) ~1.00 bytes per char
// "€" (U+20AC)  ~2.00 bytes per char

What’s happening

Each call builds a 10-million-character string and measures how much the heap grew after a full GC. Concatenation creates a "rope" (a tree of pieces) rather than one block of characters, so build copies it into a single flat string and lets the rope be collected — otherwise you'd be measuring the rope.
"a" and "é" both fit in Latin-1, so V8 uses its one-byte representation: about 1 byte per character.
"€" is U+20AC, above 255, so the string needs the two-byte representation: about 2 bytes per character.
keep holds the strings so the next measurement's gc() doesn't free them. Exact numbers are V8 internals that can change; the point is that "two bytes per character" isn't a rule.

The durable reason to prefer Buffer is meaning, not size: use Buffers for bytes (files you pass through, binary protocols, crypto, compression) and decode to a string only when the data is text and you need to work with it as text.

#What a stream is

fs.readFile waits until the entire file is in memory and then gives it to you in one piece. A stream gives you the data in chunks as it becomes available, and lets you process each chunk and let it go before the next arrives. The difference is like filling a bathtub before you can use any water, versus drinking from a tap: memory use is bounded by the chunk size and a small internal buffer, not by the size of the data, and work can start as soon as the first chunk arrives.

Buffer and stream are different ideas that work together: a buffer is bytes in memory; a stream is data moving over time, usually delivered as a series of Buffer chunks.

To see the difference, make a 200 MiB test file:

make-big-file.cjs

const fs = require("node:fs");

const line = "The quick brown fox jumps over the lazy dog 0123456789\n";
const oneMiB = Buffer.from(line.repeat(Math.ceil((1024 * 1024) / line.length)).slice(0, 1024 * 1024));
const fd = fs.openSync("big.txt", "w");
for (let i = 0; i < 200; i++) fs.writeSync(fd, oneMiB);
fs.closeSync(fd);
console.log(fs.statSync("big.txt").size); // 209715200

What’s happening

It builds exactly 1 MiB of text once, then writes it 200 times with synchronous writes — fine in a one-off script, never in a server.
The result is 209,715,200 bytes. The later examples read it.

AdvancedMeasured: peak memory of readFile hashing

Now hash it both ways and compare the peak memory of the process (maxRSS):

hash-readfile.cjs

const fs = require("node:fs");
const crypto = require("node:crypto");

fs.readFile("big.txt", (err, data) => {
  if (err) throw err;
  const hash = crypto.createHash("sha256").update(data).digest("hex");
  console.log("readFile:", hash.slice(0, 12), "| peak RSS", Math.round(process.resourceUsage().maxRSS / 1024), "MB");
});
// readFile: 00ed6a34339a | peak RSS 246 MB

AdvancedMeasured: peak memory of stream hashing

hash-stream.cjs

const fs = require("node:fs");
const crypto = require("node:crypto");

const hash = crypto.createHash("sha256");
let chunks = 0;
fs.createReadStream("big.txt")
  .on("data", (chunk) => {
    chunks++;
    hash.update(chunk);
  })
  .on("end", () => {
    console.log("stream:", hash.digest("hex").slice(0, 12), "| chunks", chunks, "| peak RSS", Math.round(process.resourceUsage().maxRSS / 1024), "MB");
  });
// stream: 00ed6a34339a | chunks 3200 | peak RSS 87 MB

What’s happening

readFile allocates one 200 MiB Buffer, fills it, and only then calls back. Peak RSS 246 MB: an empty Node process is about 41 MB on this machine, plus the whole file.
createReadStream reads 64 KiB at a time (its default highWaterMark). Each chunk is passed to the data handler, fed into the hash, and then unreferenced, so the garbage collector can reuse that memory.
200 MiB / 64 KiB = 3200 chunks. Same hash, so the same bytes were processed.
Peak RSS 87 MB, and it would stay about the same for a 20 GB file. With readFile, a 5 GB file would need 5 GB of memory — and readFile refuses anything over 2 GiB anyway (on Node 26.7 a 3 GB file fails with ERR_FS_FILE_TOO_LARGE).
Streaming bounds memory only if you let chunks go. Pushing every chunk into an array "to process later" rebuilds the readFile problem by hand.

AdvancedStreaming a file from an HTTP server

The same thing matters most in servers, where many requests run at once. HTTP requests are readable streams and responses are writable, so a big file can go straight from disk to socket:

serve-file.cjs

const http = require("node:http");
const fs = require("node:fs");
const { pipeline } = require("node:stream");

const server = http.createServer((req, res) => {
  res.setHeader("Content-Type", "text/plain");
  pipeline(fs.createReadStream("big.txt"), res, (err) => {
    if (err) console.log("server: pipeline failed —", err.code);
  });
});

server.listen(0, () => {
  const { port } = server.address();
  http.get(`http://127.0.0.1:${port}/`, (res) => {
    let received = 0;
    res.on("data", (c) => (received += c.length));
    res.on("end", () => {
      console.log("client: got", received, "bytes |", res.headers["transfer-encoding"]);
      console.log("server peak RSS", Math.round(process.resourceUsage().maxRSS / 1024), "MB");
      // A second client that hangs up after the first chunk:
      http.get(`http://127.0.0.1:${port}/`, (res2) => {
        res2.once("data", () => res2.destroy());
        res2.on("close", () => setTimeout(() => server.close(), 50));
      });
    });
  });
});

// client: got 209715200 bytes | chunked
// server peak RSS 90 MB
// server: pipeline failed — ERR_STREAM_PREMATURE_CLOSE

What’s happening

Each request gets its own read stream, connected to res with pipeline. No Content-Length was set, so Node uses Transfer-Encoding: chunked and sends data as it's read.
The client receives all 209,715,200 bytes while the process (server and client here) peaks at about 90 MB. A readFile-based handler would hold a 200 MiB copy per concurrent request.
The second client destroys its socket after the first chunk. The server's res closes early, so pipeline destroys the file stream (closing the file descriptor) and reports ERR_STREAM_PREMATURE_CLOSE. Without that cleanup, every aborted download would leak an open file.
In Express, res.sendFile and res.download do this streaming for you.

AdvancedLimits: internal buffering and blocking handlers

Streams aren't magic, though. They still buffer internally (up to the highWaterMark), and a slow synchronous computation inside a data handler still blocks the event loop — streams fix memory and flow, not CPU (see Blocking the event loop).

#The four kinds of streams

Every Node stream is one of four types, all EventEmitters from node:stream:

Type	Data flows	Core examples
Readable	out of it, to you	`fs.createReadStream`, `http.IncomingMessage` (`req` on the server), `process.stdin`
Writable	from you, into it	`fs.createWriteStream`, the server's `res`, `process.stdout`
Duplex	both ways, independently	`net.Socket` (a TCP connection), `tls.TLSSocket`
Transform	in, then out — output computed from input	`zlib.createGzip()`, `crypto.createCipheriv()`

A Duplex has two unrelated sides: what you write to a socket goes to the other machine, and what you read comes from it. A Transform is a Duplex whose readable side is derived from its writable side. (One nuance: the HTTP server's res implements the writable interface — write() returning a boolean, drain, finish — but on Node 26.7 it extends the older base Stream class, so res instanceof Writable is false. pipeline and pipe accept it all the same.)

Implementing each type takes one method, plus an error-first callback where the type consumes data:

four-kinds.cjs

const { Readable, Writable, Duplex, Transform } = require("node:stream");

// Readable: read() is called whenever the consumer wants more data.
let n = 0;
const numbers = new Readable({
  read() {
    n++;
    this.push(n <= 3 ? `chunk${n} ` : null); // null = end of data
  },
});

// Writable: write() receives each chunk; callback() means "ready for the next one".
const sink = new Writable({
  write(chunk, encoding, callback) {
    console.log("sink got", JSON.stringify(chunk.toString()), Buffer.isBuffer(chunk));
    callback();
  },
});

numbers.pipe(sink).on("finish", () => {
  // Object mode: any JS value is a chunk; highWaterMark counts objects, not bytes.
  const users = Readable.from([{ name: "Rohit" }, { name: "Neha" }]);
  console.log("objectMode:", users.readableObjectMode);
  users.on("data", (u) => console.log("object chunk:", u));

  // Transform: output is computed from input.
  const upper = new Transform({
    transform(chunk, encoding, callback) {
      callback(null, chunk.toString().toUpperCase());
    },
  });
  upper.on("data", (c) => console.log("transform out:", c.toString()));
  upper.write("hello");

  // Duplex: two independent sides, like a socket.
  const echoLater = new Duplex({
    read() {},
    write(chunk, encoding, callback) {
      setTimeout(() => this.push(`reply to ${chunk}`), 10);
      callback();
    },
  });
  echoLater.on("data", (c) => console.log("duplex read side:", c.toString()));
  echoLater.write("ping");
});

// sink got "chunk1 " true
// sink got "chunk2 " true
// sink got "chunk3 " true
// objectMode: true
// transform out: HELLO
// object chunk: { name: 'Rohit' }
// object chunk: { name: 'Neha' }
// duplex read side: reply to ping

What’s happening

numbers.pipe(sink) starts the flow. The pipe asks numbers for data, which calls read(); it pushes "chunk1 ", and the pipe passes it to sink.write. This repeats until read() pushes null, which means end of stream; then the pipe calls sink.end() and sink emits finish.
The sink received Buffers (true), not the strings that were pushed: byte streams convert strings to Buffers by default. callback() tells the Writable this chunk is done; until it's called, the next chunk waits in the buffer — this is where backpressure comes from.
Readable.from(array) makes an object-mode stream: each element, even an object, is one chunk, and the highWaterMark counts objects (16 by default for object-mode streams you construct; Readable.from uses 1).
The Transform's transform calls callback(null, output) — error-first: null for no error, then the data to push to the readable side. upper.write("hello") produces HELLO on the readable side.
The Duplex's sides are separate: writing "ping" doesn't automatically produce output; the write handler decides to push a reply 10 ms later. A TCP socket works the same way, with the network in between.
transform out prints before the object chunks because each stream schedules its own data events on later ticks; the two streams don't wait for each other.

AdvancedWhich stream types you'll actually write

You rarely subclass Duplex yourself; you'll mostly use core streams and occasionally write a Transform (see Transform streams) or an async generator (see Streams as async iterables).

#Stream events and lifecycle

Streams report their progress through events, and knowing the order is what lets you clean up correctly. For a readable: data for each chunk, end when there's no more data, then close when the underlying resource (the file descriptor) is released. For a writable: drain when its buffer has emptied after backpressure, finish when end() was called and every chunk has been handed to the destination, then close. error can arrive instead, and is followed by close.

lifecycle.cjs

const fs = require("node:fs");

// small.txt contains "line one\nline two\n" (18 bytes)
const rs = fs.createReadStream("small.txt", { highWaterMark: 8 });
const ws = fs.createWriteStream("copy.txt");
for (const ev of ["open", "ready", "end", "close", "error"]) rs.on(ev, () => console.log(`read : ${ev}`));
for (const ev of ["open", "ready", "drain", "finish", "close", "error"]) ws.on(ev, () => console.log(`write: ${ev}`));

console.log("flowing before data listener:", rs.readableFlowing);
rs.on("data", (chunk) => {
  console.log(`read : data ${JSON.stringify(chunk.toString())} → write() returned ${ws.write(chunk)}`);
});
console.log("flowing after data listener:", rs.readableFlowing);
rs.on("end", () => ws.end());

// flowing before data listener: null
// flowing after data listener: true
// read : open
// read : ready
// write: open
// write: ready
// read : data "line one" → write() returned true
// read : data "\nline tw" → write() returned true
// read : data "o\n" → write() returned true
// read : end
// write: finish
// read : close
// write: close

What’s happening

Creating a file stream starts opening the file in the background. Nothing has been read; readableFlowing is null, meaning "no consumer has chosen a mode yet".
Adding a data listener switches the stream into flowing mode (readableFlowing becomes true): it will read and emit chunks as fast as it can.
open (with the file descriptor) and ready fire for both files. Then three data events, each up to 8 bytes because of highWaterMark: 8: "line one", "\nline tw", "o\n" — 8 + 8 + 2 = 18. Chunks are cut by size, not by line; splitting into lines is your job (see Transform streams).
Each ws.write returns true: 8 bytes is far below the write stream's 64 KiB highWaterMark, so there's no backpressure and no drain.
end means the readable has delivered everything. Our handler calls ws.end(), so once its buffered writes complete, the writable emits finish. Then both streams release their file descriptors and emit close. Use finish for "all my data was written", and close for "the resource is released".

Advancederror replaces the normal ending

When something goes wrong, error replaces the normal ending:

stream-error.cjs

const fs = require("node:fs");

const rs = fs.createReadStream("does-not-exist.txt");
for (const ev of ["open", "end", "close"]) rs.on(ev, () => console.log(`read : ${ev}`));
rs.on("error", (err) => console.log("read : error", err.code, "| destroyed:", rs.destroyed));

// read : error ENOENT | destroyed: true
// read : close

What’s happening

The open fails with ENOENT, so the stream emits error — and not open or end. Code that only listens for end to clean up never runs.
Node streams auto-destroy on error: destroyed is true and close follows.
Streams are EventEmitters, so an error event with no listener is thrown and crashes the process (see The error event). Every stream you create needs an error handler — or use pipeline, which attaches them for you.

AdvancedFlowing vs paused mode

A readable is in one of two modes. Flowing pushes data at you via data events. Paused (the alternative) makes you pull with read(), typically inside a readable event:

paused-mode.cjs

const fs = require("node:fs");

const rs = fs.createReadStream("small.txt", { highWaterMark: 8 });
rs.on("readable", () => {
  let chunk;
  while ((chunk = rs.read(4)) !== null) console.log("read(4) →", JSON.stringify(chunk.toString()));
});
rs.on("end", () => console.log("end, flowing =", rs.readableFlowing));

// read(4) → "line"
// read(4) → " one"
// read(4) → "\nlin"
// read(4) → "e tw"
// read(4) → "o\n"
// end, flowing = false

What’s happening

Adding a readable listener keeps the stream paused (readableFlowing ends up false). Node fills the internal buffer and emits readable to say "there's data — come and get it".
rs.read(4) takes exactly 4 bytes from the buffer, regardless of how the file was chunked, and returns null when fewer than 4 are buffered (so the loop stops and waits for the next readable).
At the end of the file, read(4) returns the last 2 bytes "o\n" — once the stream has ended, it hands over whatever is left.
Paused mode is useful for parsers that need a fixed number of bytes (a 4-byte length header, then the body). Most code uses flowing mode, pipeline or for await. rs.pause() and rs.resume() switch a flowing stream off and on, which is how manual backpressure works.

#Backpressure

Backpressure is what happens when a producer is faster than its consumer: reading from an SSD at 2 GB/s and uploading over a 10 MB/s link, say. Something has to give. Either memory fills with chunks waiting to be written, or the producer slows down. Streams build the slowing-down in.

The mechanism is on the writable side. Every writable has a buffer and a highWaterMark threshold (64 KiB by default for byte streams on Node 26.7). writable.write(chunk) always accepts the chunk, then returns true if the buffered amount is still below the threshold, or false if it has reached it. false means "I've taken this one, but stop sending until I emit drain". It's like a water level marker on a tank: you're told when the water passes the line, and you're expected to stop pouring.

Here's the signal in its simplest form — a writable that takes 100 ms per chunk, with a 16-byte highWaterMark:

backpressure-signal.cjs

const { Writable } = require("node:stream");
const start = Date.now();

const slow = new Writable({
  highWaterMark: 16, // bytes
  write(chunk, encoding, callback) {
    setTimeout(callback, 100); // pretend the disk/network takes 100 ms per chunk
  },
});
slow.on("drain", () => console.log(`drain at ${Date.now() - start} ms, buffered=${slow.writableLength}`));

for (let i = 1; i <= 4; i++) {
  const ok = slow.write(Buffer.alloc(6, "x")); // 6 bytes each
  console.log(`write #${i} returned ${ok}  buffered=${slow.writableLength}  needDrain=${slow.writableNeedDrain}`);
}

// write #1 returned true  buffered=6  needDrain=false
// write #2 returned true  buffered=12  needDrain=false
// write #3 returned false  buffered=18  needDrain=true
// write #4 returned false  buffered=24  needDrain=true
// drain at 403 ms, buffered=0

What’s happening

write #1: the stream is idle, so chunk 1 goes straight to write() and is "in progress" for 100 ms. writableLength counts in-progress plus queued bytes: 6. 6 < 16, so it returns true.
write #2: chunk 1 is still in progress, so chunk 2 is queued: 12 buffered. 12 < 16 → true.
write #3: 18 buffered, which is not below 16, so it returns false and sets writableNeedDrain. The chunk was still accepted — don't write it again.
write #4 ignores the signal and writes anyway: 24 buffered, false again. Nothing stops you; write() never refuses data. That's why ignoring the return value is so dangerous.
The chunks complete at 100, 200, 300 and 400 ms. When the buffer empties after a false, the stream emits drain: at ~400 ms, with writableLength back to 0.

AdvancedRespecting false and drain

A producer that respects the signal stops at false and continues on drain:

respect-drain.cjs

const { Writable } = require("node:stream");
const start = Date.now();
const t = () => `${String(Date.now() - start).padStart(4)} ms`;

const slow = new Writable({
  highWaterMark: 16,
  write(chunk, encoding, callback) {
    setTimeout(callback, 100);
  },
});

let i = 0;
function produce() {
  while (i < 6) {
    i++;
    const ok = slow.write(Buffer.alloc(6, "x"));
    console.log(t(), `write #${i} → ${ok}  buffered=${slow.writableLength}`);
    if (!ok) {
      console.log(t(), "  pausing until drain");
      slow.once("drain", () => {
        console.log(t(), "  drain");
        produce();
      });
      return;
    }
  }
  slow.end();
}
slow.on("finish", () => console.log(t(), "finish"));
produce();

//    0 ms write #1 → true  buffered=6
//    1 ms write #2 → true  buffered=12
//    1 ms write #3 → false  buffered=18
//    1 ms   pausing until drain
//  303 ms   drain
//  304 ms write #4 → true  buffered=6
//  304 ms write #5 → true  buffered=12
//  304 ms write #6 → false  buffered=18
//  304 ms   pausing until drain
//  606 ms   drain
//  607 ms finish

What’s happening

0 ms: writes 1–3 go in; the third returns false with 18 bytes buffered. produce registers a one-time drain listener and returns — the loop stops. The buffer never holds more than 18 bytes.
~300 ms: the three chunks have been written (100 ms each), the buffer is empty, drain fires, and produce() picks up where it left off (i is a shared counter).
~304 ms: writes 4–6; the sixth returns false again, so it waits once more.
~606 ms: drain; produce() finds i === 6, leaves the loop and calls end(). Once nothing is left to write, finish fires.
once matters: with on, every pause would add another permanent listener and produce would eventually run several times per drain.

AdvancedMeasured: ignoring backpressure at scale

At this scale it looks academic, so here's what happens with real sizes — 5000 chunks of 64 KiB (≈312 MB) into a writable that needs about 1 ms per chunk:

ignore-backpressure.cjs

const { Writable } = require("node:stream");
const mb = (n) => `${(n / 1024 / 1024).toFixed(1)} MB`;

const slow = new Writable({
  write(chunk, encoding, callback) {
    setTimeout(callback, 1);
  },
});

const chunk = Buffer.alloc(64 * 1024);
let falses = 0;
for (let i = 0; i < 5000; i++) {
  if (!slow.write(Buffer.from(chunk))) falses++; // ignore the return value and keep going
}
console.log("write() returned false", falses, "times");
console.log("buffered in the writable:", mb(slow.writableLength));
console.log("RSS:", mb(process.memoryUsage().rss));
slow.end();

// write() returned false 5000 times
// buffered in the writable: 312.5 MB
// RSS: 364.0 MB

AdvancedMeasured: respecting backpressure at scale

respect-backpressure.cjs

const { Writable } = require("node:stream");
const { once } = require("node:events");

const slow = new Writable({
  write(chunk, encoding, callback) {
    setTimeout(callback, 1);
  },
});

(async () => {
  const chunk = Buffer.alloc(64 * 1024);
  let maxBuffered = 0;
  for (let i = 0; i < 5000; i++) {
    if (!slow.write(Buffer.from(chunk))) {
      maxBuffered = Math.max(maxBuffered, slow.writableLength);
      await once(slow, "drain"); // wait for the buffer to empty
    }
  }
  slow.end();
  await once(slow, "finish");
  console.log("max buffered", maxBuffered / 1024, "KiB | peak RSS", Math.round(process.resourceUsage().maxRSS / 1024), "MB");
})();

// max buffered 64 KiB | peak RSS 73 MB   (70–91 MB across runs)

What’s happening

In the first version the loop is synchronous, so the writable can't make any progress while it runs: all 5000 chunks pile up in its buffer. write() returned false every time (each chunk is already 64 KiB, the whole highWaterMark), and the loop ignored it.
Result: 312.5 MB sitting in the writable's buffer and an RSS of 364 MB, for a destination that could never go faster than ~1 chunk per ms. With a 20 GB input the process would run out of memory.
In the second version, a false makes the loop await once(slow, "drain"). The producer is suspended until the consumer catches up, so at most one chunk (64 KiB) is ever waiting.
The total time is about the same (6.3 s vs 6.4 s when I timed both) — the destination is the bottleneck either way — but peak RSS is 70–90 MB (it varies with when the GC runs), and it stays flat however much data you send. Backpressure doesn't make anything faster; it makes memory use independent of input size.
pipe, pipeline and for await all do this waiting for you, which is the main reason to prefer them over hand-written data handlers.

AdvancedCorrecting my notes: manual pause and resume

Correcting my notes

My notes had a hand-rolled version with readable.pause() on false and readable.resume() on drain. The backpressure part is right, but the snippet never calls writable.end(). Running it on big.txt: all 209,715,200 bytes are written and the reader ends, but the writable's finish never fires and writableEnded is still false at exit — the file descriptor stays open until the process dies, which in a long-running server is a leak. It also has no error handlers, so a read or write error crashes the process. The fixed version:

manual-backpressure.cjs

const fs = require("node:fs");

const readable = fs.createReadStream("big.txt", { highWaterMark: 64 * 1024 });
const writable = fs.createWriteStream("destination.txt");

readable.on("data", (chunk) => {
  if (!writable.write(chunk)) {
    readable.pause(); // stop reading until the writer catches up
    writable.once("drain", () => readable.resume());
  }
});
readable.on("end", () => writable.end()); // the missing line
readable.on("error", (err) => writable.destroy(err));
writable.on("error", (err) => {
  readable.destroy();
  console.error("copy failed:", err.message);
});
writable.on("finish", () => console.log("copy finished:", writable.bytesWritten, "bytes"));
// copy finished: 209715200 bytes

What’s happening

Each chunk is written; on false, reading pauses and resumes on the next drain — the same backpressure logic as the notes.
end on the reader calls writable.end(), so the writer flushes, emits finish, and closes its file.
An error on either side destroys the other, so no file descriptor is left open, and the error is reported instead of crashing the process.
Five lines of plumbing for one copy — which is exactly what pipeline(readable, writable, callback) does in one call. Hand-written versions are for understanding; use pipeline in real code.

AdvancedReadable side: push returning false

Readables have the matching signal on their side: inside a custom read(), this.push(chunk) returns false when the readable's own buffer has reached its highWaterMark, which means "stop pushing until read() is called again". Core streams like fs.createReadStream already obey it — which is why the file isn't read faster than you consume it.

#pipe vs pipeline

readable.pipe(writable) connects two streams and handles backpressure: it pauses the source when write() returns false and resumes it on drain, and calls writable.end() when the source ends. What it doesn't do is deal with errors. An error in one stream isn't passed to the others, and nothing is destroyed, so file descriptors and sockets can be left open.

stream.pipeline(...streams, callback) — or the promise version from node:stream/promises — does everything pipe does, plus error handling and cleanup: if any stream fails or closes early, it destroys every stream in the chain and reports one error.

pipe-vs-pipeline.cjs

const { Readable, Writable, pipeline } = require("node:stream");

function makeStreams() {
  let i = 0;
  const source = new Readable({
    read() {
      i++;
      if (i === 3) this.destroy(new Error("source failed mid-way"));
      else this.push(`chunk${i}`);
    },
  });
  const dest = new Writable({
    write(chunk, encoding, callback) {
      callback();
    },
  });
  return { source, dest };
}

const a = makeStreams();
a.source.pipe(a.dest);
a.source.on("error", (err) => console.log("pipe: source error →", err.message));
a.dest.on("close", () => console.log("pipe: dest closed")); // never fires

setTimeout(() => {
  console.log("pipe: dest destroyed?", a.dest.destroyed, "| ended?", a.dest.writableEnded);

  const b = makeStreams();
  pipeline(b.source, b.dest, (err) => {
    console.log("pipeline callback →", err?.message);
    console.log("pipeline: dest destroyed?", b.dest.destroyed);
  });
}, 50);

// pipe: source error → source failed mid-way
// pipe: dest destroyed? false | ended? false
// pipeline callback → source failed mid-way
// pipeline: dest destroyed? true

What’s happening

Each source emits chunk1, chunk2, then destroys itself with an error on the third read() — like a file read that fails halfway or a client that disconnects.
With pipe, the source's error reaches only the source's own error listener (which you must add, or the process crashes). The destination is simply unpiped: 50 ms later it's neither destroyed nor ended, and close never fires. With a file write stream, that's a leaked file descriptor and a half-written file nobody knows about.
With pipeline, the same failure destroys every stream in the chain (dest destroyed? true) and calls the callback once with the error. One place to handle failure, nothing left open.
A chain like a.pipe(gzip).pipe(b) needs an error listener on all three streams to be safe, and still needs manual cleanup. pipeline(a, gzip, b, cb) needs one callback.

AdvancedPromise pipeline: gzipping a large file

The promise form reads best in async code. Compressing the 200 MiB file:

gzip.cjs

const fs = require("node:fs");
const zlib = require("node:zlib");
const { pipeline } = require("node:stream/promises");

async function compress() {
  const start = Date.now();
  await pipeline(
    fs.createReadStream("big.txt"),
    zlib.createGzip(),
    fs.createWriteStream("big.txt.gz"),
  );
  console.log("Compression complete in", Date.now() - start, "ms");
  console.log(fs.statSync("big.txt").size, "→", fs.statSync("big.txt.gz").size, "bytes");
  console.log("peak RSS", Math.round(process.resourceUsage().maxRSS / 1024), "MB");
}
compress().catch(console.error);

// Compression complete in 380 ms   (380–460 ms across runs)
// 209715200 → 611037 bytes
// peak RSS 68 MB

What’s happening

Data flows big.txt → read stream → gzip Transform → write stream → big.txt.gz. The read stream emits 64 KiB chunks; gzip compresses them on the libuv thread pool and emits compressed chunks; the write stream writes them.
Backpressure works across the whole chain: if the disk write falls behind, gzip's output buffer fills, gzip stops accepting input, and the read stream stops reading. Memory stays at about 70 MB for a 200 MiB input.
await pipeline(...) fulfils once the last stream has finished — the file is completely written — and rejects if any stage fails, after destroying all of them. So .catch(console.error) sees any error from any stage.
The ratio (200 MiB → 0.6 MB) is extreme only because the test file repeats one line over and over; real text compresses far less.

AdvancedGenerator stages and AbortSignal in pipeline

pipeline also accepts async generator functions as stages, and an AbortSignal to cancel the whole chain:

pipeline-generator.mjs

import fs from "node:fs";
import { pipeline } from "node:stream/promises";

let lines = 0;
await pipeline(
  fs.createReadStream("big.txt", { encoding: "utf8" }),
  async function* (source) {
    let tail = "";
    for await (const chunk of source) {
      const parts = (tail + chunk).split("\n");
      tail = parts.pop(); // the last piece may be an incomplete line
      for (const line of parts) {
        lines++;
        if (lines % 1_000_000 === 0) yield `${lines}\n`;
      }
    }
  },
  fs.createWriteStream("milestones.txt"),
);
console.log(lines, "lines;", fs.readFileSync("milestones.txt", "utf8").trim().split("\n"));

const ac = new AbortController();
setTimeout(() => ac.abort(), 20);
try {
  await pipeline(fs.createReadStream("big.txt"), fs.createWriteStream("partial.txt"), { signal: ac.signal });
} catch (err) {
  console.log("aborted:", err.name, "| bytes copied before abort:", fs.statSync("partial.txt").size);
}

// 3813000 lines; [ '1000000', '2000000', '3000000' ]
// aborted: AbortError | bytes copied before abort: 32440320   (varies per run)

What’s happening

{ encoding: "utf8" } makes the read stream decode chunks into strings with a StringDecoder, so multi-byte characters are never split.
The generator receives the previous stage as source (an async iterable) and its yields become the input of the next stage. Here it counts lines, keeping the partial last line of each chunk in tail, and only emits a line every million lines.
Backpressure still applies: for await pulls a chunk only when the generator is ready, and a yield doesn't resume until the write stream has accepted the value.
The count is 3,813,000 complete lines (the very last line of the file is cut off mid-way, so it stays in tail), and milestones.txt holds three numbers.
The second pipeline is aborted after 20 ms: every stream is destroyed and the promise rejects with an AbortError. About 30 MB had been copied (the exact amount varies) — partial.txt is incomplete, so in real code you'd delete it in the catch.

AdvancedChoosing pipeline, finished or pipe

#Transform streams

Added

A Transform sits in the middle of a pipeline: it receives chunks on its writable side and pushes results to its readable side — gzip, encryption, parsing, filtering, formatting. You implement transform(chunk, encoding, callback), called once per input chunk, and optionally flush(callback), called once after the input has ended and before the output ends — the place to emit anything you've been holding back.

The simplest transform maps each chunk to a new one:

uppercase.cjs

const { Transform } = require("node:stream");
const { pipeline } = require("node:stream/promises");

const upper = new Transform({
  transform(chunk, encoding, callback) {
    callback(null, chunk.toString().toUpperCase());
  },
});

pipeline(process.stdin, upper, process.stdout).catch(console.error);
// $ printf 'hello streams\n' | node uppercase.cjs
// HELLO STREAMS

What’s happening

process.stdin is a readable; each chunk of piped input arrives at transform as a Buffer.
callback(null, output) pushes the uppercased string to the readable side and tells the stream it's ready for the next chunk. Calling callback(err) instead would fail the whole pipeline.
pipeline connects stdin → transform → stdout with backpressure and error handling; when stdin ends, the transform ends, and the output is flushed.
One flaw: chunk.toString() decodes each chunk on its own, so a multi-byte character split across two chunks would be corrupted (see Why use Buffers instead of strings). For short piped input it never shows; the next example handles it properly.

AdvancedStateful transform: a CSV parser

Most real transforms need state across chunks, because chunk boundaries fall wherever they like. A CSV parser has to cope with a row split between chunks — and, while it's at it, with a character split between chunks. It also turns bytes into objects, so its readable side is in object mode:

csv-parser.cjs

const { Readable, Transform, Writable } = require("node:stream");
const { pipeline } = require("node:stream/promises");
const { StringDecoder } = require("node:string_decoder");

class CsvParser extends Transform {
  #decoder = new StringDecoder("utf8");
  #tail = "";
  #headers = null;

  constructor() {
    super({ readableObjectMode: true }); // bytes in, objects out
  }
  _transform(chunk, encoding, callback) {
    const lines = (this.#tail + this.#decoder.write(chunk)).split("\n");
    this.#tail = lines.pop(); // the last piece may be an incomplete row
    for (const line of lines) this.#emitRow(line);
    callback();
  }
  _flush(callback) {
    const rest = this.#tail + this.#decoder.end();
    if (rest) this.#emitRow(rest); // input ended without a trailing newline
    callback();
  }
  #emitRow(line) {
    if (!line.trim()) return;
    const cells = line.split(",");
    if (!this.#headers) {
      this.#headers = cells;
      return;
    }
    this.push(Object.fromEntries(this.#headers.map((h, i) => [h, cells[i]])));
  }
}

// The chunk boundaries deliberately cut rows (and the "ë") in half.
const bytes = Buffer.from("name,city\nRohit,Toronto\nZoë,Delhi\nRomeo,Pune");
const cuts = [0, 19, 27, 40, bytes.length];
const chunks = cuts.slice(1).map((end, i) => bytes.subarray(cuts[i], end));
console.log(chunks.map((c) => c.toString()));

const rows = [];
pipeline(
  Readable.from(chunks),
  new CsvParser(),
  new Writable({
    objectMode: true,
    write(row, encoding, callback) {
      rows.push(row);
      callback();
    },
  }),
).then(() => console.log(rows));

// [ 'name,city\nRohit,Tor', 'onto\nZo�', '�,Delhi\nRomeo', ',Pune' ]
// [
//   { name: 'Rohit', city: 'Toronto' },
//   { name: 'Zoë', city: 'Delhi' },
//   { name: 'Romeo', city: 'Pune' }
// ]

What’s happening

The four test chunks are cut at byte offsets 19, 27 and 40. Printing them on their own shows the problem the parser has to solve: rows are cut in half (Rohit,Tor / onto), and offset 27 falls inside the two-byte ë, so each half prints as �.
Chunk 1 ("name,city\nRohit,Tor"): the decoder returns it unchanged. Splitting on \n gives ["name,city", "Rohit,Tor"]. The last piece might be incomplete, so it's kept in #tail; "name,city" is complete and becomes the headers. No row is pushed yet.
Chunk 2 (onto\nZo plus the first byte of ë): the decoder returns "onto\nZo" and holds the lone 0xc3 byte back. #tail + text is "Rohit,Toronto\nZo", so Rohit,Toronto is complete → push { name: 'Rohit', city: 'Toronto' }; the new tail is "Zo".
Chunk 3 starts with the second byte of ë. The decoder joins it with the held byte and returns "ë,Delhi\nRomeo", so the text is "Zoë,Delhi\nRomeo" → push Zoë's row, tail "Romeo". Without the StringDecoder, the name would have been Zo��.
Chunk 4 (",Pune") has no newline, so nothing is complete: the tail becomes "Romeo,Pune". Then the input ends and _flush runs: it emits the leftover tail as the last row. Without _flush, the final row would be silently lost — the most common bug in hand-written parsers.
readableObjectMode: true makes only the output side object mode: it accepts bytes and pushes objects, and the object-mode Writable receives one row per write. (Real CSV has quoted fields with commas and newlines inside them — use a library such as csv-parse for that; this shows the streaming mechanics.)

AdvancedGenerator stage vs Transform class

An async generator stage in pipeline (like the line counter in pipe vs pipeline) can do the same job with less ceremony. A Transform class is worth it when you want a reusable, configurable stream object, or need to control highWaterMark and object mode on each side separately.

#Copying a file with streams

The stream way to copy a file is the pipeline you've already seen: read stream → write stream. But if you're only copying, with no transformation, Node has better tools: fs.copyFile hands the whole job to the operating system in one call, and fs.cp copies whole directory trees.

copy-file.mjs

import fs from "node:fs";
import { copyFile, cp, constants, rm, readdir } from "node:fs/promises";
import { pipeline } from "node:stream/promises";

async function time(label, fn) {
  const start = performance.now();
  await fn();
  console.log(label.padEnd(24), `${(performance.now() - start).toFixed(0)} ms`);
}

await time("pipeline (streams)", () => pipeline(fs.createReadStream("big.txt"), fs.createWriteStream("copy-stream.txt")));
await time("fs.copyFile", () => copyFile("big.txt", "copy-copyfile.txt"));
await time("fs.copyFile + FICLONE", () => copyFile("big.txt", "copy-clone.txt", constants.COPYFILE_FICLONE));

try {
  await copyFile("big.txt", "copy-copyfile.txt", constants.COPYFILE_EXCL); // fail if it exists
} catch (err) {
  console.log("COPYFILE_EXCL:", err.code);
}

// tree/ contains a.txt and sub/b.txt
await rm("tree-copy", { recursive: true, force: true });
await cp("tree", "tree-copy", { recursive: true });
console.log("cp:", await readdir("tree-copy", { recursive: true }));

// pipeline (streams)       71 ms
// fs.copyFile              55 ms
// fs.copyFile + FICLONE    54 ms
// COPYFILE_EXCL: EEXIST
// cp: [ 'a.txt', 'sub', 'sub/b.txt' ]

What’s happening

The pipeline copy moves 3200 chunks of 64 KiB through JavaScript: each is read into a Buffer on the thread pool, handed to the event loop, then written by another thread-pool request. About 70 ms for 200 MiB on this Mac (an SSD, with the file in the OS cache).
fs.copyFile makes one libuv request and lets the OS copy the data without it ever becoming a JavaScript Buffer: about 55 ms here, and less CPU on the main thread.
COPYFILE_FICLONE asks for a copy-on-write clone on filesystems that support it (APFS, Btrfs, XFS): the new file shares the original's disk blocks until one of them is modified. If the filesystem can't, it silently falls back to a normal copy. (COPYFILE_FICLONE_FORCE fails instead of falling back.)
COPYFILE_EXCL refuses to overwrite: the destination exists, so it rejects with EEXIST — an atomic "copy only if new" check.
fs.cp(src, dest, { recursive: true }) copies a directory tree, like cp -r. Timings vary between runs (across runs I saw 70–140 ms for the stream copy and 55–70 ms for copyFile); the order is what's consistent.

AdvancedWhen streams are the right way to copy

So when are streams the right way to copy?

When you transform on the way (compress, encrypt, hash, filter lines).
When the source or destination isn't a local file: an upload to S3, an HTTP response, a socket, a child process.
When you want progress — count bytes in a pass-through stage — or the ability to cancel with an AbortSignal part-way.

For a plain local copy, fs.copyFile is simpler and faster.

#Streams as async iterables

Every Readable is also an async iterable: for await (const chunk of readable) reads it chunk by chunk, and the loop does the pausing for you. It's the simplest correct way to consume a stream, because backpressure, completion and errors all map onto ordinary control flow: the loop body is the consumer, the loop ending is end, and a stream error is thrown out of the loop where try/catch can catch it. (The language feature itself: Async iterators and for await…of; async generators on the Node side: Async generators and for await.)

iterate-chunks.mjs

import fs from "node:fs";

let bytes = 0;
let chunks = 0;
for await (const chunk of fs.createReadStream("big.txt")) {
  bytes += chunk.length;
  chunks++;
}
console.log("chunks", chunks, "bytes", bytes); // chunks 3200 bytes 209715200

const rs = fs.createReadStream("big.txt");
for await (const chunk of rs) break;
console.log("after break, destroyed:", rs.destroyed); // after break, destroyed: true

try {
  for await (const chunk of fs.createReadStream("missing.bin"));
} catch (err) {
  console.log("for await caught", err.code); // for await caught ENOENT
}

What’s happening

for await calls the stream's [Symbol.asyncIterator]() and then repeatedly awaits next(). Each value is one 64 KiB Buffer: 3200 of them, 209,715,200 bytes.
break calls the iterator's return(), which destroys the stream — the file is closed even though it wasn't read to the end. Same for return or a throw inside the loop.
A stream error (here ENOENT when opening) makes the pending next() reject, so it's thrown at the for await line and lands in an ordinary catch — no error listener needed.
The empty loop body (for await (...);) is valid syntax: it just drains the stream.

AdvancedFree backpressure in for await

The loop also gives you backpressure for free. With a deliberately slow loop body:

for-await-backpressure.mjs

import fs from "node:fs";
import { setTimeout as sleep } from "node:timers/promises";

const rs = fs.createReadStream("big.txt"); // 64 KiB chunks, 64 KiB highWaterMark
let consumed = 0;
for await (const chunk of rs) {
  consumed += chunk.length;
  await sleep(10); // slow consumer
  if (consumed >= 10 * 65536) break;
}
console.log("consumed", consumed / 1024, "KiB | read from disk", rs.bytesRead / 1024, "KiB | buffered", rs.readableLength / 1024, "KiB");
// consumed 640 KiB | read from disk 704 KiB | buffered 64 KiB

What’s happening

Each iteration takes at least 10 ms, much slower than the SSD can deliver.
After 10 chunks (640 KiB) have been processed, the stream has read only 704 KiB from disk: one chunk ahead, sitting in its internal buffer (readableLength = 64 KiB = the highWaterMark).
The read stream only reads when its buffer is below the highWaterMark, and the buffer only empties when the loop asks for the next chunk. So the disk reads at the speed of the loop body — memory use is one or two chunks no matter how big the file is.

AdvancedReading lines with readline

For text by line, node:readline wraps a stream and yields complete lines, handling chunk boundaries and both \n and \r\n endings:

read-lines.mjs

import fs from "node:fs";
import readline from "node:readline";

// access.log: "GET /a 200\nGET /b 500\r\nPOST /c 201\nGET /d 500" (no newline at the end)
const rl = readline.createInterface({ input: fs.createReadStream("access.log"), crlfDelay: Infinity });

let errors = 0;
for await (const line of rl) {
  console.log(JSON.stringify(line));
  if (line.endsWith(" 500")) errors++;
}
console.log("500s:", errors);

// "GET /a 200"
// "GET /b 500"
// "POST /c 201"
// "GET /d 500"
// 500s: 2

What’s happening

createInterface reads the stream, decodes it as UTF-8, splits on line endings and emits line events; iterating it with for await turns those into loop values.
The second line ends with \r\n (Windows style). crlfDelay: Infinity tells readline to treat \r\n as one line break even if the \r and \n arrive in different chunks, so no empty line and no stray \r appear.
The last line has no trailing newline, and readline still yields it once the input ends — the flush problem from Transform streams, solved for you.
Only the current line and a little buffered data are in memory, so this scans a multi-gigabyte log in constant memory. Two of the four requests were 500s.

AdvancedReadable.from: iterables into streams

Going the other way, Readable.from(iterable) turns any iterable or async iterable — an array, a generator, an async generator — into a readable stream, which you can then pipeline anywhere:

readable-from.mjs

import { Readable } from "node:stream";

async function* rowsFromDb() {
  for (let id = 1; id <= 3; id++) {
    await new Promise((r) => setTimeout(r, 5)); // pretend to fetch a row
    yield { id };
  }
}

const rows = Readable.from(rowsFromDb());
console.log("objectMode:", rows.readableObjectMode); // objectMode: true
for await (const row of rows) console.log(row);
// { id: 1 }
// { id: 2 }
// { id: 3 }

console.log(await Readable.from(["a", "b"]).map((s) => s.toUpperCase()).toArray()); // [ 'A', 'B' ]
console.log(await Readable.from("abc").toArray()); // [ 'abc' ]

What’s happening

Readable.from(rowsFromDb()) wraps the async generator. The stream is in object mode by default, so each yielded object is one chunk.
The stream pulls from the generator only when its buffer has room, so the generator isn't run ahead of the consumer — the same pull-based backpressure, end to end.
Readables also have array-like helpers (map, filter, take, toArray, …) that return new streams or promises. toArray() collects everything, so use it only for small streams.
The gotcha: Readable.from("abc") emits the whole string as one chunk, 'abc' — not three characters. Strings and Buffers are deliberately not iterated element by element.

AdvancedSummary: consume, produce and connect streams

To finish the picture: for await is the easiest way to consume a stream, async generators (alone or as pipeline stages) are the easiest way to produce or transform one, and pipeline is what connects them with error handling. Between them they replace most of the hand-written data/end/drain event code in older Node examples.