Events & EventEmitter

import { EventEmitter } from 'node:events';

class Library extends EventEmitter {
  #books = new Map();

  add(book) {
    this.#books.set(book.id, book);
    this.emit('book:added', book);
    return this;
  }
}

const library = new Library();

library.on('book:added', (book) => console.log('indexer saw', book.title));
library.once('book:added', (book) => console.log('first book ever:', book.title));
library.prependListener('book:added', () => console.log('audit log runs first'));

library.add({ id: 1, title: 'Dune' }).add({ id: 2, title: 'Emma' });

console.log(library.eventNames());                // [ 'book:added' ]
console.log(library.listenerCount('book:added')); // 2 (the once() listener is gone)
console.log(library.emit('nobody:listens'));      // false
console.log(library.emit('book:added', { id: 3, title: 'Ulysses' })); // true

audit log runs first
indexer saw Dune
first book ever: Dune
audit log runs first
indexer saw Emma
[ 'book:added' ]
2
false
audit log runs first
indexer saw Ulysses
true

import { EventEmitter } from 'node:events';

const e = new EventEmitter();
const a = () => { console.log('a'); e.off('x', b); };
const b = () => console.log('b');
const c = () => console.log('c');
e.on('x', a); e.on('x', b); e.on('x', c);

e.emit('x');           // listeners are snapshotted: b still runs this time
console.log('--');
e.emit('x');           // b is gone now

e.once('y', c);
console.log(e.listeners('y')[0] === c);        // true: the original function
console.log(e.rawListeners('y')[0] === c);     // false: the once() wrapper
console.log(e.rawListeners('y')[0].listener === c);

e.on('newListener', (name, fn) => console.log('newListener', String(name)));
e.on('removeListener', (name, fn) => console.log('removeListener', String(name)));
e.on('z', c);
e.off('z', c);
e.removeAllListeners('x');
console.log(e.eventNames());

a
b
c
--
a
c
true
false
true
newListener removeListener
newListener z
removeListener z
removeListener x
removeListener x
[ 'y', 'newListener', 'removeListener' ]

import { EventEmitter } from 'node:events';

const emitter = new EventEmitter();

emitter.on('tick', function () {
  console.log('regular function: this === emitter', this === emitter);
});
emitter.on('tick', () => {
  console.log('arrow function: this is', typeof this);
});
emitter.on('tick', () => {
  setImmediate(() => console.log('3. deferred listener work'));
});

console.log('1. before emit');
emitter.emit('tick');
console.log('2. after emit (all listeners already ran)');

1. before emit
regular function: this === emitter true
arrow function: this is undefined
2. after emit (all listeners already ran)
3. deferred listener work

import { EventEmitter, once } from 'node:events';

// Pitfall 1: emitting from a constructor happens before anyone can listen
class Importer extends EventEmitter {
  constructor() {
    super();
    this.emit('ready');                          // nobody is subscribed yet
  }
}
new Importer().on('ready', () => console.log('never printed'));

// Fix: defer the emission until the caller has subscribed
class SafeImporter extends EventEmitter {
  constructor() {
    super();
    queueMicrotask(() => this.emit('ready'));
  }
}
new SafeImporter().on('ready', () => console.log('SafeImporter: ready received'));

// Pitfall 2: two awaited once() calls in sequence can miss the second event
const ee = new EventEmitter();
async function listen() {
  await once(ee, 'first');
  console.log('got first');
  await once(ee, 'second');                      // registered one microtask too late
  console.log('got second');
}
listen();
ee.emit('first');
ee.emit('second');                               // emitted before the second once() exists
setTimeout(() => console.log('second was missed:', ee.listenerCount('second') === 1), 10);

// Fix: register both promises before emitting
const ee2 = new EventEmitter();
const both = Promise.all([once(ee2, 'first'), once(ee2, 'second')]);
ee2.emit('first');
ee2.emit('second');
console.log('Promise.all saw', (await both).length, 'events');

SafeImporter: ready received
got first
Promise.all saw 2 events
second was missed: true

import { EventEmitter, getMaxListeners, setMaxListeners, defaultMaxListeners } from 'node:events';

console.log(defaultMaxListeners);                 // 10
process.on('warning', (w) => {
  console.log(w.name, w.type, w.count, w.emitter === emitter);
  console.log(w.message);
});

const emitter = new EventEmitter();
for (let i = 0; i < 11; i++) emitter.on('data', () => {});

const quiet = new EventEmitter();
quiet.setMaxListeners(0);                         // 0 and Infinity mean unlimited
for (let i = 0; i < 50; i++) quiet.on('data', () => {});
console.log(quiet.getMaxListeners());

const ac = new AbortController();
console.log(getMaxListeners(ac.signal));          // 0: signals never warn by default
setMaxListeners(20, ac.signal, emitter);          // several targets at once
console.log(getMaxListeners(ac.signal), emitter.getMaxListeners());

10
0
0
20 20
(node:46904) MaxListenersExceededWarning: Possible EventEmitter memory leak detected. 11 data listeners added to [EventEmitter]. MaxListeners is 10. Use emitter.setMaxListeners() to increase limit
(Use `node --trace-warnings ...` to show where the warning was created)
MaxListenersExceededWarning data 11 true
Possible EventEmitter memory leak detected. 11 data listeners added to [EventEmitter]. MaxListeners is 10. Use emitter.setMaxListeners() to increase limit

import { EventEmitter, errorMonitor } from 'node:events';

const emitter = new EventEmitter();
emitter.on(errorMonitor, (err) => console.log('monitor sees:', err instanceof Error ? err.message : err));

try {
  emitter.emit('error', new Error('disk full'));
} catch (err) {
  console.log('thrown to the emit() caller:', err.message);
}

try {
  emitter.emit('error', 'just a string');
} catch (err) {
  console.log(err.name, err.code);
  console.log(err.context);
}

emitter.on('error', (err) => console.log('handled:', err.message));
emitter.emit('error', new Error('disk full again'));

monitor sees: disk full
thrown to the emit() caller: disk full
monitor sees: just a string
Error ERR_UNHANDLED_ERROR
just a string
monitor sees: disk full again
handled: disk full again

import { EventEmitter, captureRejectionSymbol } from 'node:events';

const unsafe = new EventEmitter();
unsafe.on('job', async () => { throw new Error('async listener failed'); });

const safe = new EventEmitter({ captureRejections: true });
safe.on('job', async () => { throw new Error('captured failure'); });
safe.on('error', (err) => console.log('routed to error:', err.message));
safe.emit('job');

class Worker extends EventEmitter {
  constructor() {
    super({ captureRejections: true });
  }
  [captureRejectionSymbol](err, eventName, ...args) {
    console.log(`custom handler: ${err.message} while handling "${String(eventName)}"`, args);
  }
}
const worker = new Worker();
worker.on('job', async (id) => { throw new Error('boom'); });
worker.emit('job', 42);
console.log(captureRejectionSymbol === Symbol.for('nodejs.rejection'));

process.on('unhandledRejection', (reason) => console.log('unhandledRejection:', reason.message));
unsafe.emit('job');

true
routed to error: captured failure
custom handler: boom while handling "job" [ 42 ]
unhandledRejection: async listener failed

import { EventEmitter, once, on } from 'node:events';
import { setTimeout as sleep } from 'node:timers/promises';

const importer = new EventEmitter();

// once(): a promise for the next occurrence; the resolved value is the argument ARRAY
setTimeout(() => importer.emit('ready', 'csv', 120), 10);
const [format, rows] = await once(importer, 'ready');
console.log('ready:', format, rows);

// once() rejects if 'error' is emitted first, and removes its listeners
setTimeout(() => importer.emit('error', new Error('parse failed')), 10);
try {
  await once(importer, 'ready');
} catch (err) {
  console.log('rejected:', err.message);
}
console.log('listeners left:', importer.listenerCount('ready'), importer.listenerCount('error'));

// once() with a signal
const ac = new AbortController();
setTimeout(() => ac.abort(), 10);
try {
  await once(importer, 'ready', { signal: ac.signal });
} catch (err) {
  console.log(err.name, err.code);
}

// on(): an async iterator over every occurrence, ended by a close event
queueMicrotask(async () => {
  for (const n of [1, 2, 3]) {
    importer.emit('row', `row-${n}`);
    await sleep(5);
  }
  importer.emit('end');
});
for await (const [row] of on(importer, 'row', { close: ['end'] })) {
  console.log('got', row);
}
console.log('iteration finished, listeners:', importer.listenerCount('row'));

// on() with a signal ends the loop with an AbortError
const ac2 = new AbortController();
setTimeout(() => ac2.abort(), 20);
try {
  for await (const [row] of on(importer, 'never', { signal: ac2.signal })) {
    console.log(row);
  }
} catch (err) {
  console.log('loop ended with', err.name);
}

ready: csv 120
rejected: parse failed
listeners left: 0 0
AbortError ABORT_ERR
got row-1
got row-2
got row-3
iteration finished, listeners: 0
loop ended with AbortError

import { EventEmitter, on } from 'node:events';
import { Readable } from 'node:stream';
import { setTimeout as sleep } from 'node:timers/promises';

// burst of events + slow consumer: on() buffers; highWaterMark pauses a Readable source
const ee = new EventEmitter();
setImmediate(() => {
  for (let i = 0; i < 5; i++) ee.emit('tick', i);
  ee.emit('done');
});
const seen = [];
for await (const [n] of on(ee, 'tick', { close: ['done'] })) {
  await sleep(1);
  seen.push(n);
}
console.log('slow consumer still saw all events in order:', seen.join(','));

// Readable emits 'data' and implements pause()/resume(), so highWaterMark applies
const source = Readable.from(['a', 'b', 'c', 'd']);
for await (const [chunk] of on(source, 'data', { close: ['end'], highWaterMark: 1, lowWaterMark: 1 })) {
  console.log('chunk', chunk);
}

slow consumer still saw all events in order: 0,1,2,3,4
chunk a
chunk b
chunk c
chunk d

const target = new EventTarget();

function onPing(event) {
  console.log(event.type, event instanceof Event, event.target === target);
}
target.addEventListener('ping', onPing);
target.addEventListener('ping', onPing);        // same function twice: registered once
target.addEventListener('ping', { handleEvent(e) { console.log('object listener', this !== undefined); } }, { once: true });
target.dispatchEvent(new Event('ping'));
target.dispatchEvent(new Event('ping'));

const ac = new AbortController();
target.addEventListener('data', (e) => console.log('data', e.detail), { signal: ac.signal });
target.dispatchEvent(new CustomEvent('data', { detail: { isbn: '978-1098145187' } }));
ac.abort();                                      // removes the listener
console.log(target.dispatchEvent(new CustomEvent('data', { detail: 'ignored' })));

// cancelable events: dispatchEvent returns false if preventDefault() was called
const cancelable = new Event('save', { cancelable: true });
target.addEventListener('save', (e) => e.preventDefault());
console.log(target.dispatchEvent(cancelable), cancelable.defaultPrevented);

// a throwing listener does not stop the dispatch and becomes an uncaughtException
process.on('uncaughtException', (err, origin) => console.log('uncaught:', err.message, origin));
target.addEventListener('boom', () => { throw new Error('listener threw'); });
target.addEventListener('boom', () => console.log('second listener still ran'));
target.dispatchEvent(new Event('boom'));

// async listeners: a rejection is captured the same way
target.addEventListener('later', async () => { throw new Error('async listener rejected'); });
target.dispatchEvent(new Event('later'));

ping true true
object listener true
ping true true
data { isbn: '978-1098145187' }
true
false true
second listener still ran
uncaught: listener threw uncaughtException
uncaught: async listener rejected uncaughtException

import { addAbortListener } from 'node:events';

const ac = new AbortController();
// Any code holding the signal can silence later 'abort' listeners
ac.signal.addEventListener('abort', (e) => e.stopImmediatePropagation());
ac.signal.addEventListener('abort', () => console.log('addEventListener listener: skipped'));
addAbortListener(ac.signal, () => console.log('addAbortListener listener: still runs'));
ac.abort();

addAbortListener listener: still runs

import { addAbortListener, getEventListeners } from 'node:events';

const ac = new AbortController();
const disposable = addAbortListener(ac.signal, (event) => {
  console.log('aborted, reason:', ac.signal.reason.message, event.type);
});
console.log(getEventListeners(ac.signal, 'abort').length);   // 1
ac.abort(new Error('user cancelled'));
disposable[Symbol.dispose]();                                  // removes the listener

// A listener added after the signal is already aborted still runs, in a microtask
addAbortListener(ac.signal, () => console.log('late listener runs anyway'));
console.log('end of script');

1
aborted, reason: user cancelled abort
end of script
late listener runs anyway

import { addAbortListener, getEventListeners } from 'node:events';

const ac = new AbortController();
{
  using _listener = addAbortListener(ac.signal, () => console.log('never runs'));
  console.log('inside scope:', getEventListeners(ac.signal, 'abort').length);
}
console.log('after scope:', getEventListeners(ac.signal, 'abort').length);

inside scope: 1
after scope: 0

import { createServer } from 'node:http';
import { Readable } from 'node:stream';
import { spawn } from 'node:child_process';
import { once } from 'node:events';

// http.Server is an EventEmitter
const server = createServer((req, res) => res.end('ok'));
server.on('listening', () => console.log('server listening'));
server.on('close', () => console.log('server closed'));
server.listen(0);
await once(server, 'listening');
const { port } = server.address();

// the ClientRequest and IncomingMessage objects emit events as well
const res = await fetch(`http://127.0.0.1:${port}/`);
console.log('fetch status', res.status);
server.close();
server.closeAllConnections();
await once(server, 'close');

// streams are EventEmitters
const readable = Readable.from(['a', 'b']);
readable.on('data', (chunk) => console.log('data', chunk));
await once(readable, 'end');
console.log('readable ended');

// child processes: 'exit' fires when the process ends, 'close' when its stdio streams are closed too
const child = spawn(process.execPath, ['-e', 'console.log("hi from child")']);
child.stdout.on('data', (d) => process.stdout.write(`child said: ${d}`));
child.on('exit', (code) => console.log('exit', code));
child.on('close', (code) => console.log('close', code));
await once(child, 'close');

// the process object is an EventEmitter too
process.on('exit', (code) => console.log('process exit', code));
console.log(process instanceof (await import('node:events')).EventEmitter);

server listening
fetch status 200
server closed
data a
data b
readable ended
child said: hi from child
exit 0
close 0
true
process exit 0

import { EventEmitter } from 'node:events';

type BookEvents = {
  added: [id: number, title: string];
  removed: [id: number];
  error: [error: Error];
};

class Catalog extends EventEmitter<BookEvents> {
  add(id: number, title: string): void {
    this.emit('added', id, title);
  }
}

const catalog = new Catalog();
catalog.on('added', (id, title) => console.log(id, title.toUpperCase()));
catalog.add(1, 'Dune');

1 DUNE

import { EventEmitterAsyncResource } from 'node:events';
import { AsyncLocalStorage } from 'node:async_hooks';

const als = new AsyncLocalStorage();

// An ordinary emitter runs listeners in the context of whoever calls emit()
// EventEmitterAsyncResource runs them in the context captured at construction
const plain = new (await import('node:events')).EventEmitter();
let tracked;
als.run('req-1', () => {
  tracked = new EventEmitterAsyncResource({ name: 'BookshelfEmitter' });
});

plain.on('x', () => console.log('plain emitter listener sees:', als.getStore()));
tracked.on('x', () => console.log('tracked emitter listener sees:', als.getStore()));

als.run('req-2', () => {
  plain.emit('x');
  tracked.emit('x');
});
tracked.emitDestroy();

plain emitter listener sees: req-2
tracked emitter listener sees: req-1