Event Emitter

The names data, end, and drain from the streams lesson all come from a single mechanism. The same mechanism is used in the HTTP server, in child processes, and in the process object’s signal notifications: most of the runtime’s asynchronous interfaces are built on top of this one class.

This lesson covers that class and turns the measurement collector’s data source into an event-based one: a source that announces when a record arrives, warns when a threshold is crossed, and reports a broken line.

The Built-in Implementation of the Observer Pattern

The EventEmitter class in the node:events module keeps a listener list per name. Listeners are added to the list with on, and the list is called in order with emit. This is a direct application of the higher-order function concept from the Programming Fundamentals course: behavior is stored as a value and called when its time comes.

The pattern’s gain is that it loosens the coupling between producer and consumer. The source does not know who is listening; the listener does not know the source’s inner workings. The only contract between them is the event name and the shape of the payload.

// emitter.mjs
import { EventEmitter } from 'node:events';

class MeasurementSource extends EventEmitter {
  feed(line) {
    try {
      const record = JSON.parse(line);
      this.emit('measurement', record);
      if (record.metric === 'temperature' && record.value > 24) {
        this.emit('threshold-exceeded', record);
      }
    } catch (error) {
      this.emit('bad-line', { line, error });
    }
  }
}

const source = new MeasurementSource();

source.on('measurement', (r) => console.log('measurement :', r.node, r.metric, r.value));
source.once('threshold-exceeded', (r) => console.log('threshold (once):', r.node, r.value));
source.on('bad-line', ({ line }) => console.log('bad line    :', line));

source.feed('{"node":"edge-01","metric":"temperature","value":21.4}');
source.feed('{"node":"edge-03","metric":"temperature","value":24.1}');
source.feed('{"node":"edge-03","metric":"temperature","value":24.6}');
source.feed('broken');

console.log('measurement listener count:', source.listenerCount('measurement'));
console.log('threshold listener count :', source.listenerCount('threshold-exceeded'));

node emitter.mjs

measurement : edge-01 temperature 21.4
measurement : edge-03 temperature 24.1
threshold (once): edge-03 24.1
measurement : edge-03 temperature 24.6
bad line    : broken
measurement listener count: 1
threshold listener count : 0

A listener added with once is removed from the list after the first emission; the zero on the output’s last line confirms this. This is also why no warning was printed on the second threshold crossing. on is used for a persistent listener, once for a one-time one; when a listener needs to be removed by hand, the off call has to be given the listener’s exact same reference — a listener added as an anonymous function expression can never be removed.

Emission Is Synchronous

Because the name is “event,” it might be assumed to enter a queue. It does not: emit calls listeners right then, in the order they were added, and returns once all of them are done.

// sync.mjs
import { EventEmitter } from 'node:events';
const e = new EventEmitter();
e.on('event', () => console.log('2 listener'));
console.log('1 before emit');
e.emit('event');
console.log('3 after emit');

node sync.mjs

1 before emit
2 listener
3 after emit

This has two consequences. First, long synchronous work done inside a listener stops the event loop like everything else; emitting an event does not defer the work. Second, an error thrown inside a listener propagates to wherever the emit call was made — into the middle of source code that never expected it.

If a listener needs to defer its work to free the caller, the deferral is written explicitly: the work is handed to a timer or a microtask. The source itself cannot make that decision.

The Privilege of the Error Event

The name error is handled specially inside the class. If an emission is made under this name while there is no listener, the payload is thrown, and if not caught, the process terminates.

// errors.mjs
import { EventEmitter } from 'node:events';

const withListener = new EventEmitter();
withListener.on('error', (e) => console.log('caught:', e.message));
withListener.emit('error', new Error('source closed'));

const withoutListener = new EventEmitter();
withoutListener.emit('error', new Error('this will be thrown'));
console.log('this line is unreachable');

Only a single line reaches standard output:

node errors.mjs 2>/dev/null

caught: source closed

The process terminates on the second emission; the line this line is unreachable is never written, and a stack trace lands on standard error. The trace’s content carries file paths and so varies by machine.

This behavior is a deliberate design. If an asynchronous source’s error is swallowed silently, the program keeps working with wrong data, and the failure shows up much later, somewhere unrelated. The rule is: every object that can emit an error gets an error listener attached. Since stream objects are event emitters too, this rule applies to them as well; one of the reasons for using pipeline in the previous lesson is that it sets up this listener automatically for every stream in the chain.

The Listener Limit

When the number of listeners added under one name crosses the default limit, the runtime writes a warning:

// limit.mjs
import { EventEmitter } from 'node:events';
const e = new EventEmitter();
for (let i = 0; i < 11; i += 1) e.on('measurement', () => {});
console.log('listener count:', e.listenerCount('measurement'));

node limit.mjs

listener count: 11
(node:80078) MaxListenersExceededWarning: Possible EventEmitter memory leak detected. 11 measurement listeners added to [EventEmitter]. MaxListeners is 10. Use emitter.setMaxListeners() to increase limit
(Use `node --trace-warnings ...` to show where the warning was created)

The number at the start of the warning line is the process id and changes with every run. The warning is not an error; the program keeps running.

The reason: if the listener list keeps growing, this usually means a new listener is added on every request and never removed. As the list grows, the values held by the closed-over functions cannot be released either, and a memory leak forms — exactly the situation described while covering reachability-based garbage collection in the Asynchronous JavaScript and the Runtime course.

If genuinely many listeners are needed, the limit is raised with setMaxListeners. Before raising the limit, why the list is growing has to be explained.

Awaiting Events as Promises

To combine the event-based interface with async/await, the module offers two helpers. The once function returns a promise that waits for a single event; the on function turns events into an asynchronous iterable.

// promise.mjs
import { EventEmitter, once, on } from 'node:events';

const source = new EventEmitter();

// Waiting for a single event: as a promise
setTimeout(() => source.emit('ready', { node: 'edge-01' }), 20);
const [payload] = await once(source, 'ready');
console.log('waited with once:', payload);

// Iterating a continuously flowing stream of events
setTimeout(() => {
  source.emit('measurement', 21.4);
  source.emit('measurement', 21.9);
  source.emit('done');
}, 20);

const controller = new AbortController();
source.once('done', () => controller.abort());

try {
  for await (const [value] of on(source, 'measurement', { signal: controller.signal })) {
    console.log('flowing event    :', value);
  }
} catch (error) {
  if (error.name !== 'AbortError') throw error;
  console.log('iteration ended  :', error.name);
}

node promise.mjs

waited with once: { node: 'edge-01' }
flowing event    : 21.4
flowing event    : 21.9
iteration ended  : AbortError

Both helpers give listener arguments as an array; for single-argument events, the first element is taken by destructuring.

The controller in the second part is the way to end an endless iteration. A loop set up with on does not end on its own — even if the source stops emitting events, the loop keeps waiting. Ending it is done with an abort signal, and the abort surfaces from the iterator as an AbortError. This is the same mechanism introduced for request cancellation in the Asynchronous JavaScript and the Runtime course.

One warning: iteration set up with on accumulates events arriving while the loop body is running. If the body is slow, this accumulation is unbounded — unlike streams, an event emitter carries no backpressure. This is the reason a stream is used instead of an event emitter for high-volume sources.

Summary

• An event emitter is the observer-pattern implementation that keeps a listener list per name; most of the runtime’s asynchronous interfaces are built on top of this class.
• emit is synchronous: it calls listeners right then, in the order they were added, and returns once all of them are done; emission does not defer work.
• The error name is privileged; an emission with no listener throws the payload and terminates the process. Every object that can emit an error gets a listener attached.
• The warning written when the listener count exceeds the limit usually reports an accumulation of un-removed listeners and the memory leak that follows from it.
• The once and on helpers turn events into a promise and an asynchronous iteration form; iteration is ended with an abort signal and carries no backpressure.

Next Step

All of the measurement collector’s pieces are ready: reading a file with a stream, building the record, summarizing, and event-based notification. The only thing missing is serving the summary outward. The next lesson builds the HTTP server: that the request is a readable stream and the response a writable one, and how the request and response structure examined line by line in the How the Internet Works course is met in code.