EventEmitter and streams

EventEmitter rules

const { EventEmitter } = require("node:events");
const bus = new EventEmitter();

bus.on("order", (o) => console.log("A", o.id));
bus.once("order", (o) => console.log("B", o.id));
bus.emit("order", { id: 1 });   // A 1, B 1   (in registration order)
bus.emit("order", { id: 2 });   // A 2        (once removed itself)

• emit is synchronous: listeners run before emit returns, in order. A throwing listener stops the rest.
• emit returns true if there were listeners.
• Emitting "error" with no listener throws. Always attach one to emitters you own.
• off(event, fn) needs the same function reference. An inline arrow can't be removed later.
• More than 10 listeners on one event prints a leak warning: usually on inside a request handler.

Building your own emitter (a build lab favourite)

class Emitter {
  #handlers = new Map();
  on(evt, fn) { (this.#handlers.get(evt) ?? this.#handlers.set(evt, []).get(evt)).push(fn); return this; }
  off(evt, fn) { this.#handlers.set(evt, (this.#handlers.get(evt) ?? []).filter((h) => h !== fn && h.orig !== fn)); return this; }
  once(evt, fn) { const w = (...a) => { this.off(evt, w); fn(...a); }; w.orig = fn; return this.on(evt, w); }
  emit(evt, ...args) {
    const hs = [...(this.#handlers.get(evt) ?? [])];   // copy: handlers may unsubscribe while we loop
    hs.forEach((h) => h(...args));
    return hs.length > 0;
  }
}

The copy in emit and w.orig (so off(evt, fn) works for once handlers) are exactly what the tests probe.

Streams in one paragraph

Streams move data in chunks instead of loading it all. writable.write() returns false when its buffer is full: wait for "drain" before writing more (that's backpressure). Prefer pipeline from node:stream/promises: it wires the stages, forwards errors and cleans up.

await pipeline(fs.createReadStream("in.csv"), transform, fs.createWriteStream("out.csv"));