EventEmitter в Node.js: on, emit, once и свои события

EventEmitter это класс из модуля events в Node, который позволяет объекту сообщать об именованных событиях. Слушателя регистрируют через emitter.on('name', fn), а вызывают через emitter.emit('name', ...args); выполняются все слушатели с этим именем, в порядке добавления, с переданными вами аргументами. HTTP-серверы, потоки и process все являются EventEmitter, поэтому одни и те же несколько методов работают в Node везде.

const EventEmitter = require('events');

const shop = new EventEmitter();

shop.on('order', (item, qty) => {
    console.log('order received:', qty, 'x', item);
});
shop.on('order', (item) => {
    console.log('update stock for', item);
});

console.log('before emit');
shop.emit('order', 'coffee', 2);
console.log('after emit');

Оба слушателя выполняются между before emit и after emit: emit вызывает их синхронно и возвращается, только когда они закончат. Поменяйте местами два вызова shop.on, и две средние строки тоже поменяются.

on, once и off

on держит слушателя, пока вы его не удалите, once удаляет его после первого вызова, а off (то же самое, что removeListener) удаляет одного слушателя. Чтобы удалить слушателя, нужна та же самая функция, которую вы добавили, поэтому дайте ей имя, а не передавайте стрелочную функцию прямо в вызов.

const EventEmitter = require('events');
const door = new EventEmitter();

function logOpen(who) {
    console.log('opened by', who);
}

door.on('open', logOpen);
door.once('open', () => console.log('first open ever'));

door.emit('open', 'Ada');
door.emit('open', 'Linus');

door.off('open', logOpen);
console.log('listeners left:', door.listenerCount('open'));
console.log('emit returned', door.emit('open', 'Grace'));

first open ever выводится один раз, после Ada. Когда logOpen удалён, никто не слушает, поэтому последний emit возвращает false; door.removeAllListeners('open') убрал бы обоих слушателей одним вызовом.

Событие error

'error' особенное. Если вы его отправите, а никто не слушает, emit выбросит ошибку, а в реальной программе неперехваченное исключение роняет процесс. Всегда добавляйте слушателя 'error' к emitter'ам, которые могут сломаться, например к потокам и серверам.

const EventEmitter = require('events');
const job = new EventEmitter();

try {
    job.emit('error', new Error('disk full'));
} catch (err) {
    console.log('emit threw:', err.message);
}

job.on('error', (err) => {
    console.log('handled:', err.message);
});
job.emit('error', new Error('disk full'));
console.log('still running');

Без слушателя emit выбрасывает ошибку, и её ловит catch; когда слушатель добавлен, тот же emit просто вызывает его, и программа продолжает работу. Любое другое событие без слушателей молча игнорируется.

Наследование от EventEmitter в своём классе

Обычно класс наследуют от EventEmitter, вызывают super() в конструкторе и this.emit, когда что-то происходит. Тогда вызывающий код подписывается на ваш объект так же, как на любой объект Node.

const EventEmitter = require('events');

class Upload extends EventEmitter {
    constructor(name, chunks) {
        super();
        this.name = name;
        this.chunks = chunks;
    }

    start() {
        for (let i = 1; i <= this.chunks; i++) {
            this.emit('progress', Math.round((i / this.chunks) * 100));
        }
        this.emit('done', this.name);
    }
}

const up = new Upload('photo.png', 4);
up.on('progress', (pct) => console.log(pct + '%'));
up.once('done', (name) => console.log(name, 'uploaded'));
up.start();

Программа выводит 25%, 50%, 75%, 100% и затем photo.png uploaded. Перенесите up.start() выше двух строк с up.on, и ничего не выведется: события срабатывают раньше, чем кто-то начал слушать.

Та же ловушка возникает, если отправлять события из конструктора. Библиотеки, которым нужно отправить событие сразу, оборачивают вызов в process.nextTick, и он выполняется после строк с on у вызывающего кода (см. страницу про event loop).

Ждём событие через events.once

events.once(emitter, name) возвращает промис, который при следующей отправке события разрешается его аргументами в виде массива. Так одноразовое событие превращается в то, что можно ждать через await, а если emitter раньше отправит 'error', промис будет отклонён.

import { EventEmitter, once } from 'events';

const db = new EventEmitter();

setTimeout(() => db.emit('ready', 'v1.4', 3), 100);

console.log('waiting for ready');
const [version, tables] = await once(db, 'ready');
console.log('ready:', version, 'with', tables, 'tables');

await ждёт, пока таймер отправит ready, а затем массив деструктурируется в два аргумента. Для события, которое срабатывает много раз, events.on(emitter, name) даёт асинхронный итератор, по которому можно пройти циклом for await.

Частые ошибки

Добавление слушателя внутри функции, которая вызывается часто. Вызов on в каждом обработчике запроса копит слушателей и память. Добавьте слушателя один раз или используйте once.

Удаление стрелочной функции, переданной прямо в вызов. off('data', () => ...) ничего не удаляет, потому что эта стрелочная функция новая. Храните ссылку на функцию, которую добавили.

Ожидание, что слушатели выполнятся позже. emit синхронный, поэтому медленный слушатель задерживает код, который отправил событие. Пометка слушателя как async не помогает: его код всё равно выполняется в том же потоке до первого await. Разбейте тяжёлую работу на части с setImmediate между ними или вынесите её в рабочий поток (worker thread).