Events and streams all 3 written

example.mjs

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18import { EventEmitter } from "node:events";

const shop = new EventEmitter();
const log = [];

shop.on("order", (item, qty) => log.push(`email: thanks for ${qty} x ${item}`));
shop.on("order", (item) => log.push(`stock: reserve ${item}`));
shop.once("order", () => log.push("analytics: first order ever"));

shop.emit("order", "tea", 2);
shop.emit("order", "cake", 1);

console.log(log);
console.log("listeners left for 'order':", shop.listenerCount("order"));

check("every listener ran, in the order it was added", log.slice(0, 3),
      ["email: thanks for 2 x tea", "stock: reserve tea", "analytics: first order ever"]);
check("once() removed itself after the first", shop.listenerCount("order"), 2);

example.mjs

1
2
3
4
5
6
7
8
9
10
11
12
13
14import { EventEmitter } from "node:events";

const bus = new EventEmitter();
const order = [];

bus.on("saved", () => order.push("listener ran"));

order.push("before emit");
bus.emit("saved");
order.push("after emit");

console.log(order);

check("the listener finished before emit returned", order, ["before emit", "listener ran", "after emit"]);

example.mjs

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17import { EventEmitter } from "node:events";

const bus = new EventEmitter();
bus.on("job", () => { throw new Error("listener failed"); });

let caught;
try {
  bus.emit("job");
} catch (err) {
  caught = err.message;
}

console.log("emit threw:", caught);
console.log("emit with no listeners returns", bus.emit("nothing-listening"));

check("a listener's error surfaces from emit", caught, "listener failed");
check("emit reports whether anyone was listening", bus.emit("nothing-listening"), false);

example.mjs

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31import { EventEmitter } from "node:events";

class Download extends EventEmitter {
  constructor(size) {
    super();
    this.size = size;
  }

  start() {
    let received = 0;
    const timer = setInterval(() => {
      received += this.size / 4;
      this.emit("progress", Math.round((received / this.size) * 100));
      if (received >= this.size) {
        clearInterval(timer);
        this.emit("done", { bytes: received });
      }
    }, 5);
    return this;
  }
}

const seen = [];
new Download(1000)
  .on("progress", (pct) => seen.push(pct))
  .on("done", (info) => {
    console.log("progress:", seen.join("% ") + "%");
    console.log("done:", info);
    check("progress was reported four times", seen, [25, 50, 75, 100]);
  })
  .start();

example.mjs

1
2
3
4
5
6
7
8
9
10
11import { EventEmitter } from "node:events";

const connection = new EventEmitter();
const problems = [];

connection.on("error", (err) => problems.push(err.message));
connection.emit("error", new Error("connection reset by peer"));

console.log("still running; problems:", problems);

check("a listener turns a crash into a handled event", problems, ["connection reset by peer"]);

example.mjs

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20import { EventEmitter, once, on } from "node:events";

const server = new EventEmitter();
setTimeout(() => server.emit("listening", 8080), 10);

const [port] = await once(server, "listening");      // a promise for the next event
console.log("listening on", port);

const feed = new EventEmitter();
setTimeout(() => ["a", "b", "c"].forEach((x) => feed.emit("item", x)), 10);

const items = [];
for await (const [item] of on(feed, "item")) {       // every event, as an async iterator
  items.push(item);
  if (items.length === 3) break;
}
console.log("items:", items);

check("once() resolves with the event's arguments", port, 8080);
check("on() turns a stream of events into a for await loop", items, ["a", "b", "c"]);

example.mjs

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16import { EventEmitter } from "node:events";

const warnings = [];
process.on("warning", (w) => warnings.push(w.name));

const bus = new EventEmitter();

// A classic leak: adding a listener every time something happens, never removing it.
for (let request = 1; request <= 11; request++) {
  bus.on("shutdown", () => {});
}

setImmediate(() => {
  console.log("listeners now:", bus.listenerCount("shutdown"));
  check("the eleventh listener triggered Node's warning", warnings, ["MaxListenersExceededWarning"]);
});

example.mjs

1
2
3
4
5
6
7
8
9
10
11
12
13import { createReadStream, writeFileSync } from "node:fs";

writeFileSync("data.bin", Buffer.alloc(1024 * 1024));    // a 1 MiB file

const sizes = [];
const stream = createReadStream("data.bin");

stream.on("data", (chunk) => sizes.push(chunk.length));
stream.on("end", () => {
  console.log(`${sizes.length} chunks, each ${sizes[0]} bytes`);
  check("the file arrived as 16 chunks of 64 KiB", sizes.length, 16);
  check("which add up to the whole file", sizes.reduce((a, b) => a + b, 0), 1024 * 1024);
});

poem.txt

1
2
3
4The fog comes
on little cat feet.
It sits looking
over harbor and city

example.mjs

1
2
3
4
5
6
7
8
9
10
11
12
13import { createReadStream } from "node:fs";

let lines = 0;
let words = 0;
for await (const chunk of createReadStream("poem.txt", { encoding: "utf8", highWaterMark: 16 })) {
  lines += chunk.split("\n").length - 1;
  words += chunk.split(/\s+/).filter(Boolean).length;
}

console.log({ lines, words }, "<- the poem has 14 words");

check("every newline was counted", lines, 4);
check("but words split across chunks were counted twice", words, 17);

example.mjs

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16import { createWriteStream } from "node:fs";
import { readFile } from "node:fs/promises";
import { once } from "node:events";

const out = createWriteStream("log.txt");

out.write("first\n");
out.write("second\n");
out.end("last\n");                    // write this, then close

await once(out, "finish");             // every byte has been handed to the OS

const text = await readFile("log.txt", "utf8");
console.log(JSON.stringify(text));

check("the writes arrived in order", text, "first\nsecond\nlast\n");

example.mjs

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21import { Transform, Readable } from "node:stream";
import { pipeline } from "node:stream/promises";
import { createWriteStream } from "node:fs";
import { readFile } from "node:fs/promises";

const upper = new Transform({
  transform(chunk, encoding, callback) {
    callback(null, chunk.toString().toUpperCase());   // pass on a changed chunk
  },
});

await pipeline(
  Readable.from(["streams ", "are ", "just ", "chunks\n"]),
  upper,
  createWriteStream("shout.txt"),
);

const result = await readFile("shout.txt", "utf8");
console.log(result.trim());

check("every chunk went through the transform", result, "STREAMS ARE JUST CHUNKS\n");

example.mjs

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20import { createReadStream, createWriteStream, writeFileSync } from "node:fs";
import { stat, readFile } from "node:fs/promises";
import { pipeline } from "node:stream/promises";
import { createGzip, createGunzip } from "node:zlib";

// A log file with a lot of repetition, as log files have.
const lines = Array.from({ length: 20_000 }, (_, i) => `2026-09-29 INFO request ${i % 50} served in 12ms`);
writeFileSync("access.log", lines.join("\n"));

await pipeline(createReadStream("access.log"), createGzip(), createWriteStream("access.log.gz"));
await pipeline(createReadStream("access.log.gz"), createGunzip(), createWriteStream("restored.log"));

const before = (await stat("access.log")).size;
const after = (await stat("access.log.gz")).size;
console.log(`${before} bytes -> ${after} bytes, ${(100 * after / before).toFixed(1)}% of the original`);

const same = (await readFile("access.log")).equals(await readFile("restored.log"));

check("gzip made it much smaller", after < before / 10, true);
check("and gunzip restored it exactly", same, true);

example.mjs

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29import { Readable, Writable } from "node:stream";
import { pipeline } from "node:stream/promises";

let written = 0;
const destination = new Writable({
  write(chunk, enc, callback) {
    written += chunk.length;
    callback();
  },
});

async function* source() {
  yield "row 1\n";
  yield "row 2\n";
  throw new Error("the database connection dropped");
}

let failure;
try {
  await pipeline(Readable.from(source()), destination);
} catch (err) {
  failure = err.message;
}

console.log("pipeline rejected with:", failure);
console.log("destination destroyed?", destination.destroyed, "after", written, "bytes");

check("the source's error reached the caller", failure, "the database connection dropped");
check("and pipeline cleaned up the destination", destination.destroyed, true);

example.mjs

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37import { Writable } from "node:stream";
import { once } from "node:events";

// A destination that takes 1 ms per chunk -- a slow disk, a slow client.
const slowDestination = () => new Writable({
  highWaterMark: 16 * 1024,
  write(chunk, encoding, callback) { setTimeout(callback, 1); },
});
const chunk = Buffer.alloc(16 * 1024);          // 16 KiB of data per write
const kib = (n) => `${(n / 1024).toFixed(0)} KiB`;

// 1. Ignoring write()'s answer.
const careless = slowDestination();
let carelessPeak = 0;
for (let i = 0; i < 500; i++) {
  careless.write(chunk);
  carelessPeak = Math.max(carelessPeak, careless.writableLength);
}
careless.end();
await once(careless, "finish");

// 2. Stopping when write() returns false, until "drain".
const careful = slowDestination();
let carefulPeak = 0;
for (let i = 0; i < 500; i++) {
  const ok = careful.write(chunk);
  carefulPeak = Math.max(carefulPeak, careful.writableLength);
  if (!ok) await once(careful, "drain");
}
careful.end();
await once(careful, "finish");

console.log("ignoring backpressure, most data waiting in memory:", kib(carelessPeak));
console.log("respecting it:                                     ", kib(carefulPeak));

check("the careless loop buffered all 500 chunks", carelessPeak, 500 * 16 * 1024);
check("the careful one never held more than one", carefulPeak <= 16 * 1024, true);

example.mjs

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21import { Readable, Writable } from "node:stream";
import { pipeline } from "node:stream/promises";

const destination = new Writable({
  highWaterMark: 16 * 1024,
  write(chunk, encoding, callback) { setTimeout(callback, 1); },
});

let peak = 0;
const sampler = setInterval(() => { peak = Math.max(peak, destination.writableLength); }, 1);

const chunk = Buffer.alloc(16 * 1024);
await pipeline(
  Readable.from((function* () { for (let i = 0; i < 300; i++) yield chunk; })()),
  destination,
);
clearInterval(sampler);

console.log(`peak waiting in the destination: ${peak / 1024} KiB`);

check("pipeline paused the source whenever the destination was full", peak <= 32 * 1024, true);

example.mjs

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39import { createReadStream, createWriteStream, statSync } from "node:fs";
import { readFile } from "node:fs/promises";
import { createInterface } from "node:readline";
import { once } from "node:events";

const mib = (n) => n / 1024 / 1024;

// Write 800,000 log lines, respecting backpressure.
const out = createWriteStream("access.log");
for (let i = 0; i < 800_000; i++) {
  const level = i % 7 === 0 ? "ERROR" : "INFO ";
  if (!out.write(`2026-09-29T10:00:00Z ${level} request ${i} took ${i % 500}ms\n`)) {
    await once(out, "drain");
  }
}
out.end();
await once(out, "finish");
console.log(`access.log is ${mib(statSync("access.log").size).toFixed(1)} MiB`);

// 1. One line at a time.
const samples = [];
let errors = 0;
let n = 0;
for await (const line of createInterface({ input: createReadStream("access.log"), crlfDelay: Infinity })) {
  if (line.includes(" ERROR ")) errors++;
  if (++n % 200_000 === 0) samples.push(mib(process.memoryUsage().heapUsed));
}
console.log("line by line:", errors, "errors; heap at each quarter:", samples.map((s) => s.toFixed(0) + " MiB").join(", "));

// 2. The whole file at once.
const before = process.memoryUsage().heapUsed;
const all = (await readFile("access.log", "utf8")).split("\n");
const grew = mib(process.memoryUsage().heapUsed - before);
const errors2 = all.filter((line) => line.includes(" ERROR ")).length;
console.log(`all at once:  ${errors2} errors; heap grew by ${grew.toFixed(0)} MiB`);

check("both counted the same errors", errors2, errors);
check("reading line by line kept the heap flat", Math.max(...samples) - Math.min(...samples) < 16, true);
check("reading it whole cost more than twice the file's size", grew > 2 * mib(statSync("access.log").size), true);