Streams

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.

// 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`);

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.

// 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)}`);
}

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:

printf '{"node":"edge-01","metric":"humidity","value":48.0}\nbroken line\n{"node":"edge-02","metric":"humidity","value":50.0}\n' > broken.ndjson

// 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);
}

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.

// 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();

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.