TopolearnStreams let you handle files and network data far larger than memory with constant memory use.
A stream delivers data in chunks over time instead of all at once. Four kinds exist:
- Readable — a source:
fs.createReadStream, an HTTP request on the server,process.stdin. - Writable — a destination:
fs.createWriteStream, an HTTP response,process.stdout. - Duplex — both, independently (a TCP socket).
- Transform — a duplex whose output is computed from its input (
zlib.createGzip(), a CSV parser).
Copying a 5 GB log with readFile + writeFile needs 5 GB of RAM; with streams it needs about 64 KB at a time.
The right way to connect streams is `pipeline` from node:stream/promises:
await pipeline(createReadStream("in.log"), createGzip(), createWriteStream("in.log.gz"));It wires each stream to the next, propagates errors from any stage, destroys all streams on failure, and resolves when everything finishes. The older a.pipe(b) does none of the error handling — a failing source leaves the destination open and leaks file descriptors.
Readable streams are async iterables, so the simplest way to consume one is for await (const chunk of stream). Chunks are Buffers unless you set an encoding (createReadStream(path, { encoding: "utf8" })) or the stream is in object mode.
To write your own transform, new Transform({ transform(chunk, enc, cb) { cb(null, result) } }) works, but you can also pass an async generator function as a pipeline stage — it receives the previous stage as an iterable and yields output. That is often the most readable option.
Node also implements web streams (ReadableStream, used by fetch bodies); Readable.fromWeb and Readable.toWeb convert between the two.
Example
import { createReadStream, createWriteStream } from "node:fs";
import { pipeline } from "node:stream/promises";
import { createGzip } from "node:zlib";
import { writeFile } from "node:fs/promises";
await writeFile("access.log", "GET /\nPOST /login\nGET /about\n".repeat(1000));
// Async-generator stage: keep only GET lines, uppercase them
async function* onlyGets(source) {
let rest = "";
for await (const chunk of source) {
const lines = (rest + chunk).split("\n");
rest = lines.pop(); // keep a partial last line
for (const line of lines) if (line.startsWith("GET")) yield line.toUpperCase() + "\n";
}
}
await pipeline(
createReadStream("access.log", { encoding: "utf8" }),
onlyGets,
createGzip(),
createWriteStream("gets.log.gz"),
);
console.log("done"); // errors in any stage reject the pipeline promiseRemember
- Streams move data in chunks with bounded memory
- Connect stages with
pipelinefromnode:stream/promises, not.pipe() - Readables are async iterables; async generators make easy transforms
Watch out for
- Using
.pipe()without error handling, leaking open files when a stage fails. - Assuming a chunk boundary aligns with a line or record boundary.