Streams

readFile gives you the whole file as one string. That is perfect for a settings file and a disaster for a 20 GB log: the program needs 20 GB of memory before it can look at the first line. A stream is the alternative: data handled one chunk at a time, in order, as it becomes available. Only a few chunks are in memory at any moment, so a stream can process a file of any size, and it can start on the first chunk while the rest is still on its way. Network connections, file reads, the output of child processes and compressed data are all streams.

There are four kinds in Node:

• a Readable produces chunks: a file being read, an HTTP request body, process.stdin;
• a Writable consumes chunks: a file being written, process.stdout;
• a Transform is both: chunks go in, changed chunks come out, as in compression;
• a Duplex has two independent sides, such as a network socket.

Streams are event emitters ("data", "end", "error", "drain", "finish"), and a Readable is also an async iterable, so async iteration is the easiest way to consume one. The examples use withTempDir from the files lesson, which creates a temporary directory and removes it at the end.

Writing a big file, and backpressure

A chunk is a Buffer (bytes) for most streams, or a string when you set an encoding. fs.createWriteStream(path) returns a Writable. You write with write(chunk) and finish with end().

One rule matters: backpressure. A fast producer and a slow consumer would make data pile up in memory. So write() returns a boolean. true means “send more”. false means “my buffer is full: stop, and wait for the drain event”. The buffer size that counts as full is the stream’s highWaterMark.

The first part of the program writes a file of 100,000 lines and honours the rule: whenever write says false, it waits for drain. once from node:events turns the event into a promise.

import { createWriteStream, createReadStream } from "node:fs";
import { createInterface } from "node:readline";
import { once } from "node:events";
import { join } from "node:path";
import { withTempDir } from "./fp_helper.mjs";

await withTempDir(async (dir) => {
  const file = join(dir, "big.txt");
  const out = createWriteStream(file);
  for (let i = 1; i <= 100000; i++) {
    if (!out.write(`line number ${i}\n`)) await once(out, "drain");
  }
  out.end();
  await once(out, "finish");
  console.log("file written");

The file is about 1.8 MB. Now read it back as a stream. A Readable is async iterable, and each iteration gives one chunk. To show that no whole-file string is needed, we count the chunks and the newline bytes (code 10) chunk by chunk.

  let chunks = 0;
  let newlines = 0;
  let largest = 0;
  for await (const chunk of createReadStream(file)) {
    chunks++;
    largest = Math.max(largest, chunk.length);
    for (const byte of chunk) if (byte === 10) newlines++;
  }
  console.log(`${chunks} chunks, largest ${largest} bytes`);
  console.log(`${newlines} newlines`);

file written
28 chunks, largest 65536 bytes
100000 newlines

The file arrived in 28 pieces of at most 65,536 bytes (64 KiB, the default highWaterMark of file streams), and never sat in memory whole. Counting bytes works because a newline is one byte. A chunk boundary can fall anywhere, even in the middle of a line or of a multi-byte character, so code that works on text needs more care. The readline module takes that care and hands you whole lines.

  let count = 0;
  let longest = "";
  const input = createReadStream(file);
  const lines = createInterface({ input, crlfDelay: Infinity });
  for await (const line of lines) {
    count++;
    if (line.length > longest.length) longest = line;
  }
  console.log(count, longest);
});

100000 line number 100000

Each line is looked at and dropped, so memory stays small however many lines there are. This is the answer to “count the lines of a very large file”: a for await over a stream, never readFile.

Readable streams from your own data

Readable.from(iterable) makes a stream out of any iterable or async iterable: an array, a generator, an async generator from the async iteration lesson. Such streams are in object mode, so each chunk can be any JavaScript value and not only bytes.

import { Readable } from "node:stream";

for await (const chunk of Readable.from(["a", "b", "c"])) {
  console.log("chunk:", chunk);
}

async function* countdown(from) {
  for (let n = from; n > 0; n--) yield { n };
}
const seen = [];
for await (const item of Readable.from(countdown(3))) seen.push(item.n);
console.log(seen);

chunk: a
chunk: b
chunk: c
[ 3, 2, 1 ]

Leaving a for await loop early with break or an exception destroys the stream and releases its resources.

Backpressure, seen up close

A tiny writable shows the numbers. It accepts chunks slowly (its write function calls callback after a millisecond) and has a highWaterMark of 4 bytes.

import { Writable } from "node:stream";
import { once } from "node:events";

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

console.log(slow.write("abc"));
console.log(slow.write("de"));
console.log(slow.writableLength, slow.writableNeedDrain);
await once(slow, "drain");
console.log("drained", slow.writableLength);

true
false
5 true
drained 0

After "abc" the buffer holds 3 bytes, below the limit of 4, so write returns true. After "de" it holds 5, so write returns false: the data was accepted, nothing was lost, but the stream asks you to pause. When the buffer has emptied, drain fires. The highWaterMark is a threshold and not a hard limit, which is why 5 bytes fit. Streams connected with pipeline, below, apply this rule for you.

Transform streams and pipeline

A Transform sits in the middle. You supply a transform(chunk, encoding, callback) function; it calls callback(error, output) when it is done with a chunk. Connecting stages by hand means wiring events and remembering to close everything on errors, and getting that wrong leaks file handles. pipeline from node:stream/promises connects the stages, applies backpressure, destroys every stream if one fails, and returns a promise that resolves when the last stage has finished.

import { Readable, Transform } from "node:stream";
import { pipeline } from "node:stream/promises";
import { createWriteStream } from "node:fs";
import { readFile } from "node:fs/promises";
import { join } from "node:path";
import { withTempDir } from "./fp_helper.mjs";

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

await withTempDir(async (dir) => {
  const target = join(dir, "out.txt");
  const source = Readable.from(["hello ", "streams\n"]);
  await pipeline(source, upper, createWriteStream(target));
  console.log(JSON.stringify(await readFile(target, "utf8")));
});

"HELLO STREAMS\n"

The source produced two chunks, the Transform changed each, and the file received the result. A byte-oriented Transform may see chunks split or joined differently from run to run, so a stage that depends on chunk boundaries is a bug. Uppercasing does not.

pipeline also accepts an async generator function as a stage, often shorter than a class: it receives the previous stage as an async iterable, and whatever it yields goes on to the next stage.

async function* exclaim(source) {
  for await (const chunk of source) yield chunk.toString().toUpperCase() + "!";
}

await withTempDir(async (dir) => {
  const target = join(dir, "out2.txt");
  await pipeline(Readable.from(["wow"]), exclaim, createWriteStream(target));
  console.log(await readFile(target, "utf8"));
});

WOW!

Errors travel through the pipeline

If any stage fails, pipeline destroys the others and rejects with that error, so one try/catch around the await handles the whole chain. These two stages use objectMode: true because they pass numbers and not bytes. The Transform fails on a negative number, and the Writable collects what reaches it.

import { Readable, Transform, Writable } from "node:stream";
import { pipeline } from "node:stream/promises";

const checker = new Transform({
  objectMode: true,
  transform(value, encoding, callback) {
    if (value < 0) return callback(new Error(`negative number: ${value}`));
    callback(null, value * 2);
  },
});
const seen = [];
const sink = new Writable({
  objectMode: true,
  write(value, encoding, callback) {
    seen.push(value);
    callback();
  },
});

Now run the stages over four numbers, one of them negative.

try {
  await pipeline(Readable.from([1, 2, -3, 4]), checker, sink);
} catch (error) {
  console.log("pipeline failed:", error.message);
}
console.log("sink saw", seen, "and destroyed:", sink.destroyed);

pipeline failed: negative number: -3
sink saw [ 2, 4 ] and destroyed: true

The first two numbers went through. The third made the Transform fail, the error reached the catch, and the sink was destroyed, so the 4 never arrived.

Compression

node:zlib provides Transform streams for gzip and others. Compressing a file is a three-stage pipeline: read, compress, write. Decompressing is the reverse. The first part writes a file of repeated text and compresses it into a.txt.gz.

import { createReadStream, createWriteStream } from "node:fs";
import { writeFile, readFile } from "node:fs/promises";
import { pipeline } from "node:stream/promises";
import { createGzip, createGunzip } from "node:zlib";
import { join } from "node:path";
import { withTempDir } from "./fp_helper.mjs";

await withTempDir(async (dir) => {
  const names = ["a.txt", "a.txt.gz", "b.txt"];
  const [plain, packed, back] = names.map((n) => join(dir, n));
  await writeFile(plain, "the same words again\n".repeat(5000));

  await pipeline(
    createReadStream(plain),
    createGzip(),
    createWriteStream(packed),
  );

The second part decompresses that file into b.txt and compares the files.

  await pipeline(
    createReadStream(packed),
    createGunzip(),
    createWriteStream(back),
  );

  const sizes = [plain, packed].map(async (f) => (await readFile(f)).length);
  const [before, after] = await Promise.all(sizes);
  console.log("smaller:", after < before / 10);
  const read = (file) => readFile(file, "utf8");
  console.log("same text:", (await read(back)) === (await read(plain)));
});

smaller: true
same text: true

The exact compressed size depends on the compression library, so the example prints comparisons and not byte counts. The repeated text shrinks to a small fraction of its size, and the round trip gives back the original.

Web streams

Browsers have their own stream classes, defined by a web standard: ReadableStream, WritableStream and TransformStream. Node provides them as globals (stable since Node 21), so code written for them runs in both places. The body of a fetch response, which you will meet in the HTTP lesson, is a ReadableStream. It is async iterable too, and pipeThrough connects a Transform.

const source = new ReadableStream({
  start(controller) {
    controller.enqueue("a");
    controller.enqueue("b");
    controller.close();
  },
});
const shout = new TransformStream({
  transform(chunk, controller) {
    controller.enqueue(chunk.toUpperCase());
  },
});

for await (const chunk of source.pipeThrough(shout)) {
  console.log(chunk);
}

A
B

The two families convert into each other: Readable.toWeb(stream) and Readable.fromWeb(webStream) (also on Writable and Duplex). For Node-only programs, classic streams and pipeline are what you will use most. For code that must run in both Node and the browser, prefer web streams.

Common pitfalls

• Using readFile for big or unknown-size data. Use a stream.
• Ignoring the return value of write(). The data piles up in memory.
• Wiring streams with .pipe() and forgetting errors. An error in one stage does not close the others. Use pipeline.
• Splitting text by chunk. A chunk may end in the middle of a line or of a character. Use readline or set the stream’s encoding.
• Forgetting objectMode when streams carry numbers or objects.
• Not waiting for finish. After end(), the data may still be on its way to the disk. Await finish, or use pipeline.
• No "error" listener on a stream you wire by hand. An unhandled "error" event throws and ends the program.

Exercises

1: alpha
2: beta
3: gamma

import { Readable } from "node:stream";
import { createInterface } from "node:readline";

const chunks = ["alpha\nbe", "ta\ngam", "ma\n"];
const lines = createInterface({ input: Readable.from(chunks) });

let number = 0;
for await (const line of lines) {
  number++;
  console.log(`${number}: ${line}`);
}

1: alpha
2: beta
3: gamma

[ 2, 4, 6 ]

import { Readable, Transform, Writable } from "node:stream";
import { pipeline } from "node:stream/promises";

const double = new Transform({
  objectMode: true,
  transform(n, encoding, callback) {
    callback(null, n * 2);
  },
});
const results = [];
const collect = new Writable({
  objectMode: true,
  write(n, encoding, callback) {
    results.push(n);
    callback();
  },
});

await pipeline(Readable.from([1, 2, 3]), double, collect);
console.log(results);

[ 2, 4, 6 ]

true true

import { Readable, Writable } from "node:stream";
import { pipeline } from "node:stream/promises";
import { createGzip, gunzipSync } from "node:zlib";

const text = "abc ".repeat(1000);
const parts = [];
const collect = new Writable({
  write(chunk, encoding, callback) {
    parts.push(chunk);
    callback();
  },
});

await pipeline(Readable.from([text]), createGzip(), collect);
const packed = Buffer.concat(parts);
console.log(gunzipSync(packed).toString() === text, packed.length < 100);

true true

Summary

• A stream handles data chunk by chunk, so memory stays small however large the data is. The kinds are Readable, Writable, Transform and Duplex.
• A Readable is async iterable: for await (const chunk of stream). Readable.from turns arrays and generators into streams, in object mode.
• Streams are event emitters. write() returns false when the buffer is full: wait for drain (backpressure), or let pipeline do it.
• pipeline from node:stream/promises connects stages, handles backpressure, cleans up on error and rejects with the failure. Async generator functions work as stages.
• readline splits a byte stream into whole lines. Never assume that chunks align with lines or characters.
• node:zlib offers gzip as Transform streams. Web streams (ReadableStream, TransformStream) are global in Node and can be converted with toWeb and fromWeb.

Next: talking to servers with fetch, and writing a small server of your own.