--- title: Streams source: 'https://academia.sh/en/courses/nodejs/streams' course: 'The Node.js Runtime' language: en updated: '2026-08-17T18:09:51+00:00' license: 'CC BY-SA 4.0' --- # Streams The four stream types, why chunk boundaries do not line up with record boundaries, object mode, building a pipeline, and measuring backpressure. The previous two lessons loaded a file into memory all at once. For a twelve-line file this costs nothing. When measurement nodes write for days, the file reaches hundreds of megabytes, and the same code demands memory equal to the whole file — then copies that memory once more to split it into lines. This lesson's question is: how do we process data without ever seeing all of it? The answer is the same as the pipeline model from the Shell Programming course — data enters at one end, passes through a few transformations, exits the other end; no step holds all of it. The runtime offers this model as first-class objects. ## Four Types A **stream** is an abstraction that carries data piece by piece. It has four types: - A **readable stream** produces data. File reading, an incoming HTTP request, a child process's standard output are of this kind. - A **writable stream** consumes data. File writing, an outgoing HTTP response, standard output are of this kind. - A **duplex stream** carries two independent directions that are both of the above at once; a network socket is like this. - A **transform stream** is a special case of duplex: it processes written data and emits it on its readable side. Compression, encryption, and parsing are of this kind. All streams are event emitters: they emit events like `data`, `end`, `error`, `drain`. The event emitter itself is the subject of a later lesson; here, higher-level tools will be used instead of listening to events directly. ## Chunk Boundaries Are Not Record Boundaries A readable stream hands out data in chunks of a size it decides on its own. This size is set with the `highWaterMark` option and has nothing to do with the line boundaries in the file. ```js // chunks.mjs import { createReadStream } from 'node:fs'; const stream = createReadStream('measurements.ndjson', { highWaterMark: 256 }); let index = 0; for await (const chunk of stream) { index += 1; console.log(`chunk ${index}: ${chunk.length} bytes, type ${chunk.constructor.name}`); } console.log(`total ${index} chunks`); ``` ```sh node chunks.mjs ``` ``` chunk 1: 256 bytes, type Buffer chunk 2: 256 bytes, type Buffer chunk 3: 256 bytes, type Buffer chunk 4: 240 bytes, type Buffer total 4 chunks ``` The file is 1008 bytes; it split into four chunks and the last one carried the remainder. None of the chunks end at a line boundary. This is the fundamental difficulty of working with streams: **establishing the record boundary is your job, not the stream's.** The `for await` loop in the example takes advantage of a readable stream being an asynchronous iterable. While the loop body runs, the stream pauses; this is the plainest form of backpressure. ## Object Mode and a Transform By default, streams carry bytes. When **object mode** is turned on, an arbitrary value is carried instead of a chunk; from some point in the pipeline onward, records flow instead of bytes. The transform below splits byte chunks into lines and keeps a leftover partial line in itself to join with the next chunk. This is the classic pattern of stream writing. ```js // summary-stream.mjs import { createReadStream } from 'node:fs'; import { Transform, Writable } from 'node:stream'; import { pipeline } from 'node:stream/promises'; // Splits byte chunks into lines; keeps a leftover partial line for the next chunk class LineSplitter extends Transform { #remainder = ''; constructor() { super({ readableObjectMode: true }); } _transform(chunk, encoding, done) { const text = this.#remainder + chunk.toString('utf8'); const lines = text.split('\n'); this.#remainder = lines.pop(); // the last piece may be partial for (const line of lines) { if (line !== '') this.push(line); } done(); } _flush(done) { if (this.#remainder !== '') this.push(this.#remainder); done(); } } // Turns lines into records and collects them bucket by bucket class SummaryCollector extends Writable { buckets = new Map(); constructor() { super({ objectMode: true }); } _write(line, encoding, done) { let record; try { record = JSON.parse(line); } catch (error) { return done(new Error(`broken line: ${line.slice(0, 30)}`, { cause: error })); } const key = `${record.node}/${record.metric}`; const bucket = this.buckets.get(key) ?? { count: 0, total: 0 }; bucket.count += 1; bucket.total += record.value; this.buckets.set(key, bucket); done(); } } const collector = new SummaryCollector(); await pipeline( createReadStream('measurements.ndjson', { highWaterMark: 256 }), new LineSplitter(), collector, ); for (const [key, { count, total }] of collector.buckets) { console.log(`${key.padEnd(18)} ${String(count).padStart(2)} measurements, average ${(total / count).toFixed(2)}`); } ``` ```sh node summary-stream.mjs ``` ``` edge-01/temperature 3 measurements, average 21.87 edge-01/humidity 2 measurements, average 47.60 edge-02/temperature 3 measurements, average 20.23 edge-02/humidity 1 measurements, average 52.50 edge-03/temperature 2 measurements, average 24.35 edge-03/humidity 1 measurements, average 41.30 ``` Three details deserve attention. `readableObjectMode` puts only the readable side into object mode; the writable side keeps accepting bytes. A transform's two sides can be configured separately, and this is exactly what this example requires. `_flush` is called once when the input ends and gives the chance to emit the leftover partial line. Without this hook, the last record of a file whose last line does not end with a newline would be lost. When an error is given to the `done` callback, the error propagates through the stream and tears down the pipeline. Using `throw` inside a writable stream does not give the same result. ## Building a Pipeline There are two ways to connect streams to each other. The `pipe` method is old and has a flaw: when a stream in the middle of the chain errors, the others do not close on their own, and open file descriptors and buffered memory are left behind. `pipeline` closes this gap. When any stream in the chain errors, it destroys all of them and reports the error from a single place. The behavior is tested below with a file containing a broken line. First create the file: ```sh printf '{"node":"edge-01","metric":"humidity","value":48.0}\nbroken line\n{"node":"edge-02","metric":"humidity","value":50.0}\n' > broken.ndjson ``` ```js // pipeline-error.mjs import { createReadStream } from 'node:fs'; import { Transform, Writable } from 'node:stream'; import { pipeline } from 'node:stream/promises'; const parser = new Transform({ readableObjectMode: true, transform(chunk, encoding, done) { for (const line of chunk.toString('utf8').split('\n')) { if (line === '') continue; try { this.push(JSON.parse(line)); } catch (error) { return done(new Error(`unparseable line: ${line}`, { cause: error })); } } done(); }, }); let counter = 0; const count = new Writable({ objectMode: true, write(record, encoding, done) { counter += 1; done(); }, }); const source = createReadStream('broken.ndjson'); try { await pipeline(source, parser, count); console.log('completed, record count:', counter); } catch (error) { console.log('pipeline error :', error.message); console.log('root cause :', error.cause.name); console.log('source destroyed:', source.destroyed); console.log('target destroyed:', count.destroyed); } ``` ```sh node pipeline-error.mjs ``` ``` pipeline error : unparseable line: broken line root cause : SyntaxError source destroyed: true target destroyed: true ``` The last two lines matter: by the time the error was reported, the source and the target had already been closed. No file descriptor leaked. The promise-returning form in the `node:stream/promises` module combines this cleanup with `try`/`catch`. ## Backpressure **Backpressure** was defined while covering queues in the Data Structures course: if the producer is faster than the consumer, the only way to stop the buffer in between from growing is to slow the producer down. The same problem also came up describing the pipeline in the Shell Programming course — there, the kernel did the slowing down. In streams, this job is entrusted to a write call's return value. A `write` call returns `false` if the data in the buffer has crossed a threshold. This is not an error, it means "slow down"; the producer waits for the `drain` event and continues. Two words will be kept separate throughout the course. **Buffer**, lowercase, is the area inside a stream holding data waiting to be processed; its size is read from the `writableLength` field. **Buffer object** — capital-B `Buffer` — is the object representing raw bytes and is the subject of the next lesson. English uses the word *buffer* for both; the distinction is kept through context here. ```js // backpressure.mjs import { Writable } from 'node:stream'; import { once } from 'node:events'; // Target that processes each record in 5 ms, buffering at most 4 records class SlowWriter extends Writable { constructor() { super({ objectMode: true, highWaterMark: 4 }); } _write(record, encoding, done) { setTimeout(done, 5); } } function records(count) { return Array.from({ length: count }, (_, i) => ({ index: i })); } // 1. Producer that ignores the return value const a = new SlowWriter(); for (const record of records(50)) a.write(record); console.log('ignoring -> records waiting in buffer:', a.writableLength); a.end(); // 2. Producer that listens to the return value const b = new SlowWriter(); let highest = 0; for (const record of records(50)) { if (!b.write(record)) { highest = Math.max(highest, b.writableLength); await once(b, 'drain'); } } console.log('listening -> highest buffer value seen:', highest); b.end(); ``` ```sh node backpressure.mjs ``` ``` ignoring -> records waiting in buffer: 50 listening -> highest buffer value seen: 4 ``` The difference between the two numbers is backpressure in its entirety. The producer ignoring the return value piled all fifty records into memory; even though the threshold was 4, the buffer rose to 50. The producer listening to the return value never let the buffer rise above 4. The production counterpart of this difference is memory consumption. A server writing fast-produced data to a slow client, if it ignores the return value, accumulates unbounded memory per client; a handful of slow clients push the process against its memory limit. The good news: `pipeline` and `pipe` do this control on their own, and so does the `for await` loop. Managing the return value by hand is only needed where you drive streams directly. There is also a built-in convenience for line-based reading: the `node:readline` module's interface turns a readable stream into a line-by-line iterable and takes on the splitting logic above. Writing your own transform is needed in formats where a record is bounded by something other than a line. ## Summary - Streams carry data piece by piece; their four types are readable, writable, duplex, and transform. - Chunk boundaries do not line up with record boundaries; establishing the record is the job of a transform that stores leftover partial data, and the `_flush` hook rescues the last record. - Object mode lets records flow instead of bytes from some point in the pipeline onward; a transform's two sides can be configured separately. - `pipeline` destroys every stream in the chain on an error and reports it from one place; `pipe` does not do this cleanup. - A `write` call returning `false` is a backpressure signal; ignored, the buffer grows without bound. `pipeline`, `pipe`, and `for await` take on this control. ## Next Step In this lesson, chunks appeared as type `Buffer`, and the encoding name was given by hand when converting to text. What happens if a chunk splits in the middle of a character was not asked. The next lesson covers the representation of raw bytes: what the buffer object is, how multi-byte characters get corrupted at a chunk boundary, and how a binary record format is encoded and decoded.