Собираем EventEmitter: on, once, off, emit и его ловушки

С подписками вы встречались в браузере: addEventListener кладёт функцию в список элемента, а событие зовёт этот список по порядку (несколько обработчиков на одном элементе). В Node то же делает класс EventEmitter из модуля node:events, только глаголы короче: on подписывает, off снимает, emit испускает событие. Объекта события нет: всё, что передали в emit после имени, слушатель получает аргументами.

Своя версия — Map «имя события → массив функций». Похожий класс уже был в веб-модуле Expo, но подписки там лежали в Set. Здесь массив, как в самом Node, а почему не Set — в заметке в конце раздела. on дописывает функцию в конец массива, off вырезает её, emit проходит по массиву и зовёт каждую. Зовёт через apply(this, args): если слушатель — обычная функция, а не стрелка, this в нём указывает на сам излучатель, как и в Node. Прогноз: в каком порядке появятся шесть строк и что вернёт emit для события, на которое никто не подписан?

class Emitter {
  events = new Map();

  on(name, fn) {
    if (!this.events.has(name)) this.events.set(name, []);
    this.events.get(name).push(fn);
    return this;
  }

  off(name, fn) {
    const list = this.events.get(name) ?? [];
    // с конца: из двух одинаковых подписок снимается последняя, как в Node
    const i = list.lastIndexOf(fn);
    if (i !== -1) list.splice(i, 1);
    return this;
  }

  emit(name, ...args) {
    const list = this.events.get(name) ?? [];
    if (list.length === 0) return false;
    for (const fn of list) fn.apply(this, args);
    return true;
  }
}

const server = new Emitter();
server.on("request", (url) => console.log("журнал:", url));
server.on("request", (url) => console.log("ответ на", url));

console.log("до emit");
console.log("emit вернул", server.emit("request", "/health"));
console.log("после emit");
console.log("close без слушателей:", server.emit("close"));

emit возвращает true, если слушатели были, и false, если нет. Событие без слушателей — не ошибка: его просто некому услышать. Почти всегда: у одного имени правило другое, до него дойдём.

Одну и ту же функцию EventEmitter разрешает подписать дважды, и позовут её дважды: браузер такой дубликат отбрасывал. Поэтому off ищет с конца — из двух одинаковых подписок снимается последняя.

Первое, что захочется дописать, — слушатель, который срабатывает один раз. Без специального метода он пишется руками: слушатель вызывает off и передаёт в него себя. Излучатель из прошлого примера лежит в подложке без изменений. У двери два слушателя: «первый гость» снимает себя, «звонок» остаётся. Прогноз: что напечатают два emit подряд?

// Emitter из прошлого примера — в подложке, без изменений
const door = new Emitter();

function firstGuest() {
  console.log("первый гость");
  door.off("open", firstGuest);
}

door.on("open", firstGuest);
door.on("open", () => console.log("звонок"));

door.emit("open");
console.log("---");
door.emit("open");

В первом проходе звонка нет, хотя его никто не снимал. Цикл for...of идёт по массиву по номерам: выполнил элемент 0 — возьмёт элемент 1. Пока выполнялся элемент 0, off его вырезал, и звонок съехал на место 0. А цикл уже шагнул на место 1, где ничего нет. Ни ошибки, ни исключения — одного слушателя просто не позвали. Так ломается любой одноразовый слушатель в наивном излучателе: сработав, он лишает события соседа.

Лечение — одна пара скобок: emit проходит не по живому массиву, а по его копии [...list] (spread в литерале массива). Копия снята до первого вызова, и что бы слушатели ни делали со списком, этот проход идёт по ней до конца. Ниже тот же класс с одной правкой в emit. Прогноз: что изменится в выводе?

class Emitter {
  events = new Map();

  on(name, fn) {
    if (!this.events.has(name)) this.events.set(name, []);
    this.events.get(name).push(fn);
    return this;
  }

  off(name, fn) {
    const list = this.events.get(name) ?? [];
    // с конца: из двух одинаковых подписок снимается последняя, как в Node
    const i = list.lastIndexOf(fn);
    if (i !== -1) list.splice(i, 1);
    return this;
  }

  emit(name, ...args) {
    const list = this.events.get(name) ?? [];
    if (list.length === 0) return false;
    for (const fn of [...list]) fn.apply(this, args);
    return true;
  }
}

const door = new Emitter();

function firstGuest() {
  console.log("первый гость");
  door.off("open", firstGuest);
}

door.on("open", firstGuest);
door.on("open", () => console.log("звонок"));

door.emit("open");
console.log("---");
door.emit("open");

Звонок звучит в обоих проходах. Правка списка теперь касается только следующих вызовов emit.

У копии есть обратная сторона, и её стоит проверить. Пусть первый слушатель снимает не себя, а соседа, и заодно подписывает третьего. Прогоним этот сценарий на своём излучателе и на EventTarget — том самом классе, что стоит за addEventListener: в браузере его можно создать и без элемента. Прогноз: что напечатает каждый?

// Emitter с копией списка — в подложке.
// Один сценарий на двух системах: первый слушатель снимает второго и подписывает третьего
function scenario(label, subscribe, unsubscribe, fire) {
  const second = () => console.log(label, "второй");
  const late = () => console.log(label, "поздний");
  subscribe(() => {
    console.log(label, "первый");
    unsubscribe(second);
    subscribe(late);
  });
  subscribe(second);
  fire();
}

const room = new Emitter();
scenario(
  "Emitter:",
  (fn) => room.on("tick", fn),
  (fn) => room.off("tick", fn),
  () => room.emit("tick"),
);

const target = new EventTarget();
scenario(
  "EventTarget:",
  (fn) => target.addEventListener("tick", fn),
  (fn) => target.removeEventListener("tick", fn),
  () => target.dispatchEvent(new Event("tick")),
);

Вывод разошёлся. Свой излучатель позвал снятого второго: копию сняли до прохода, и второй в ней был. EventTarget второго пропустил. А позднего не позвал никто.

Это два разных договора, а не случайность реализации. Спецификация DOM тоже копирует список перед проходом — потому поздний не выполнился и в браузере. Но у каждой подписки там есть отметка «снята», и перед вызовом браузер её проверяет. У EventEmitter отметки нет: что попало в копию, то и будет вызвано. Отсюда правило для кода на Node: off посреди emit этот проход не отменяет. Если слушателю нельзя работать после какого-то события — скажем, соединение уже закрыто, — он проверяет это сам, а не рассчитывает, что его успели снять.

В самом Node копия устроена хитрее, но договор тот же. По Node 24.14 и 25.8 включительно emit копировал массив при каждом вызове, как миниатюра. С 24.15 и 25.9 копии на каждый emit нет: пока идёт проход, массив помечен счётчиком, и on с off, увидев метку, правят не его, а свежую копию. Лишний массив появляется, только когда список правят посреди прохода, а снаружи разницы не видно.

Теперь одноразовую подписку можно сделать методом. once подписывает не саму функцию, а обёртку: обёртка снимает себя и зовёт настоящую. Снимает до вызова — так же браузер обходится с опцией once. Излучатель с копией списка лежит в подложке, once дописан в наследнике. Человек подписал greet через once, передумал и снял её через off. Прогноз: что напечатают два emit?

// Emitter с копией списка — в подложке
class OnceEmitter extends Emitter {
  once(name, fn) {
    const wrapper = (...args) => {
      // снять до вызова: если fn сама вызовет emit того же события, второй раз она не сработает
      this.off(name, wrapper);
      fn.apply(this, args);
    };
    return this.on(name, wrapper);
  }
}

const chat = new OnceEmitter();
function greet(who) {
  console.log("привет,", who);
}

chat.once("join", greet);
chat.off("join", greet);
console.log("emit вернул", chat.emit("join", "Аня"));
console.log("emit вернул", chat.emit("join", "Боря"));

off промахнулся, и greet сработала. В списке лежит обёртка, а lastIndexOf(greet) ищет саму greet и не находит. Второй emit вернул false, потому что обёртка, отработав, сняла себя сама. Тот, кто пишет once(greet), а потом off(greet), про обёртку не знает и знать не должен.

Поэтому обёртка хранит исходную функцию в свойстве listener, а off сравнивает искомую функцию и с элементом списка, и с тем, что элемент помнит. Так сделано и в Node: rawListeners() отдаёт обёртки, listeners() — исходные функции, и подписку через once снимает off с исходной. Прогноз: что напечатается теперь?

// Emitter с копией списка — в подложке
class OnceEmitter extends Emitter {
  once(name, fn) {
    const wrapper = (...args) => {
      this.off(name, wrapper);
      fn.apply(this, args);
    };
    wrapper.listener = fn;
    return this.on(name, wrapper);
  }

  off(name, fn) {
    const list = this.events.get(name) ?? [];
    const i = list.findLastIndex((f) => f === fn || f.listener === fn);
    if (i !== -1) list.splice(i, 1);
    return this;
  }
}

const chat = new OnceEmitter();
function greet(who) {
  console.log("привет,", who);
}

chat.once("join", greet);
console.log("в списке сама greet:", chat.events.get("join")[0] === greet);
chat.off("join", greet);
console.log("emit вернул", chat.emit("join", "Аня"));

В списке по-прежнему обёртка, но off её нашёл — по свойству listener, и emit звать больше некого. findLastIndex ищет с конца по той же причине, что lastIndexOf раньше: из двух одинаковых подписок снимается последняя.

Событие без слушателей обычно ничего не значит: emit вернул false, и всё. Одно имя — исключение. Если испущен error, а слушателей у него нет, emit бросает ту самую ошибку, которую ему передали. Проверка стоит в начале emit, до прохода по списку. В миниатюре это наследник, который после проверки зовёт родительский emit через super. Прогноз: какие строки напечатаются, а какая нет?

// Emitter с копией списка — в подложке
class StrictEmitter extends Emitter {
  emit(name, ...args) {
    if (name === "error" && !this.events.get("error")?.length) {
      throw args[0] instanceof Error ? args[0] : new Error("Unhandled error");
    }
    return super.emit(name, ...args);
  }
}

const socket = new StrictEmitter();
console.log("close без слушателей:", socket.emit("close"));

try {
  socket.emit("error", new Error("read ECONNRESET"));
  console.log("после error");
} catch (err) {
  console.log("вылетело из emit:", err.message);
}

socket.on("error", (err) => console.log("слушатель error:", err.message));
console.log("со слушателем:", socket.emit("error", new Error("read ECONNRESET")));

Строки «после error» нет: исключение вылетело из emit и перескочило её. А с подписанным слушателем то же событие стало обычным: слушатель получил ошибку, emit вернул true.

Зачем такое правило? Ошибка, которую никто не слушает, — это ошибка, которую никто не обработал. Промолчать значило бы потерять её: сокет оборвался, запрос остался без ответа, а в логе пусто. Node выбирает громкий вариант. И громкий всерьёз: error сокета испускается из колбэка, который запустил цикл событий, когда try вокруг кода, открывшего соединение, давно закрыт (где try..catch промахивается). Ловить исключение некому, и процесс завершается. А один процесс Node держит все соединения сервера разом, так что без слушателя error одно оборванное соединение роняет сервер вместе со всеми остальными. Отсюда правило: у каждого излучателя, который может испустить error, — сервера, сокета, потока — слушатель error есть всегда.

Исключение из самого слушателя тоже вылетает из emit наружу, и соседи, стоящие в списке после него, в этом проходе не выполнятся: цикл внутри emit ничего не ловит.

import { EventEmitter } from "node:events";

const socket = new EventEmitter();
socket.on("data", (chunk) => console.log("данные:", chunk));

socket.emit("data", "GET /health HTTP/1.1");
socket.emit("error", new Error("read ECONNRESET"));
console.log("после error");

данные: GET /health HTTP/1.1
file://<пример>/crash.mjs:7
socket.emit("error", new Error("read ECONNRESET"));
                     ^

Error: read ECONNRESET
    at file://<пример>/crash.mjs:7:22
    at <внутренние кадры node>
Emitted 'error' event at:
    at file://<пример>/crash.mjs:7:8
    at <внутренние кадры node>

Node.js v<версия>

Последняя ловушка сидит не в излучателе, а в том, как им пользуются. Подписка живёт, пока её не сняли, а с ней живёт всё, что слушатель замкнул. В браузере это было на подписках к document (забытая отписка и утечки). На сервере долгожитель — любой общий объект: соединение с базой, общая шина событий, сам процесс. Подписываются на него часто в обработчике запроса. Если запрос не отписался, когда ответил, каждый следующий добавляет по слушателю.

Какая подписка лишняя, Node знать не может, зато может считать. Как только у одного события одного излучателя становится больше десяти слушателей, он печатает предупреждение — не на каждой следующей подписке, а один раз. Порог по умолчанию — 10. Миниатюра ниже считает так же. Двенадцать запросов подписываются на правки в базе и не отписываются. Прогноз: где в выводе встанет предупреждение, сколько слушателей будет в конце и скольких разбудит одна правка?

// Emitter с копией списка — в подложке
class WarningEmitter extends Emitter {
  maxListeners = 10;

  on(name, fn) {
    super.on(name, fn);
    const list = this.events.get(name);
    // один раз на список: иначе предупреждение печаталось бы на каждой следующей подписке
    if (list.length > this.maxListeners && !list.warned) {
      list.warned = true;
      console.warn(`MaxListenersExceededWarning: ${list.length} ${name} listeners, MaxListeners is ${this.maxListeners}`);
    }
    return this;
  }
}

const db = new WarningEmitter();
let woken = 0;

function handleRequest(id) {
  // запросу нужна правка в базе, чтобы сбросить свой кеш
  db.on("change", () => {
    woken += 1;
  });
  return `ответ на запрос ${id}`;
}

for (let id = 1; id <= 12; id++) handleRequest(id);
console.log("слушателей change:", db.events.get("change").length);
db.emit("change");
console.log("одна правка разбудила:", woken);

Предупреждение пришло на одиннадцатом слушателе, а двенадцатый встал в список следом как ни в чём не бывало: это счётчик, а не запрет. Одна правка в базе разбудила двенадцать функций от запросов, которые давно получили ответ. На живом сервере их тысячи, каждая держит в замыкании свой запрос, и память растёт до перезапуска процесса.

Лечат отпиской, а не порогом. Запрос снимает свой слушатель, когда ответил: off там, где отправлен ответ, или once, если нужно ровно одно событие. Поднять порог через setMaxListeners(n) правильно, только когда много слушателей — норма: скажем, у комнаты чата по слушателю на участника. setMaxListeners(0) снимает счётчик совсем, а с ним и единственный сигнал, что подписки текут.

Сверим миниатюру с оригиналом. В файле ниже пять проверок статьи на EventEmitter из node:events: снятие и подписка посреди прохода, once и off с исходной функцией, error без слушателя, двенадцать подписок на одно событие. Прогноз: совпадёт ли вывод с тем, что печатали примеры, и где встанет предупреждение об утечке?

import { EventEmitter } from "node:events";

// 1 и 2: снятие и подписка посреди прохода
const room = new EventEmitter();
function first() {
  console.log("первый");
  room.off("tick", second);
  room.on("tick", late);
}
function second() {
  console.log("второй");
}
function late() {
  console.log("поздний");
}
room.on("tick", first).on("tick", second);
room.emit("tick");
console.log("слушателей tick после прохода:", room.listenerCount("tick"));

// 3: once, потом off с исходной функцией
const chat = new EventEmitter();
function greet(who) {
  console.log("привет,", who);
}
chat.once("join", greet);
console.log("в списке сама greet:", chat.rawListeners("join")[0] === greet);
chat.off("join", greet);
console.log("emit после off:", chat.emit("join", "Аня"));

// 4: error без слушателя
const socket = new EventEmitter();
console.log("close без слушателей:", socket.emit("close"));
try {
  socket.emit("error", new Error("read ECONNRESET"));
  console.log("после error");
} catch (err) {
  console.log("вылетело из emit:", err.message);
}

// 5: двенадцать подписок на одно событие
const db = new EventEmitter();
for (let id = 1; id <= 12; id++) db.on("change", () => {});
console.log("слушателей change:", db.listenerCount("change"));
console.log("конец синхронного кода");

первый
второй
слушателей tick после прохода: 2
в списке сама greet: false
emit после off: false
close без слушателей: false
вылетело из emit: read ECONNRESET
слушателей change: 12
конец синхронного кода
(node:<pid>) MaxListenersExceededWarning: Possible EventEmitter memory leak detected. 11 change listeners added to [EventEmitter]. MaxListeners is 10. Use emitter.setMaxListeners() to increase limit
(Use `node --trace-warnings ...` to show where the warning was created)

Первые девять строк — ровно то, что обещали примеры: снятый второй выполнился, поздний нет; в списке лежит обёртка, но off с greet её снял; error вылетел из emit; слушателей двенадцать. А предупреждение пришло последним, после «конец синхронного кода». Node не печатает его внутри on: он отдаёт его в process.emitWarning, а тот выводит предупреждение в поток ошибок, когда синхронный код уже закончился. (node:<pid>) в начале — номер процесса. Подсказку про --trace-warnings Node добавляет к первому предупреждению: запущенный с этим флагом, он печатает стек, и в нём видна строка с одиннадцатым on.

Чем ещё оригинал отличается от миниатюры — в таблице.

class Emitter {
  events = new Map();
  maxListeners = 10;

  on(name, fn) {
    if (!this.events.has(name)) this.events.set(name, []);
    const list = this.events.get(name);
    list.push(fn);
    if (list.length > this.maxListeners && !list.warned) {
      list.warned = true;
      console.warn(`MaxListenersExceededWarning: ${list.length} ${name} listeners, MaxListeners is ${this.maxListeners}`);
    }
    return this;
  }

  once(name, fn) {
    const wrapper = (...args) => {
      this.off(name, wrapper);
      fn.apply(this, args);
    };
    wrapper.listener = fn;
    return this.on(name, wrapper);
  }

  off(name, fn) {
    const list = this.events.get(name) ?? [];
    const i = list.findLastIndex((f) => f === fn || f.listener === fn);
    if (i !== -1) list.splice(i, 1);
    return this;
  }

  emit(name, ...args) {
    const list = this.events.get(name) ?? [];
    if (name === "error" && list.length === 0) {
      throw args[0] instanceof Error ? args[0] : new Error("Unhandled error");
    }
    if (list.length === 0) return false;
    for (const fn of [...list]) fn.apply(this, args);
    return true;
  }

  listenerCount(name) {
    return this.events.get(name)?.length ?? 0;
  }

  rawListeners(name) {
    return [...(this.events.get(name) ?? [])];
  }
}

// 1 и 2: снятие и подписка посреди прохода
const room = new Emitter();
function first() {
  console.log("первый");
  room.off("tick", second);
  room.on("tick", late);
}
function second() {
  console.log("второй");
}
function late() {
  console.log("поздний");
}
room.on("tick", first).on("tick", second);
room.emit("tick");
console.log("слушателей tick после прохода:", room.listenerCount("tick"));

// 3: once, потом off с исходной функцией
const chat = new Emitter();
function greet(who) {
  console.log("привет,", who);
}
chat.once("join", greet);
console.log("в списке сама greet:", chat.rawListeners("join")[0] === greet);
chat.off("join", greet);
console.log("emit после off:", chat.emit("join", "Аня"));

// 4: error без слушателя
const socket = new Emitter();
console.log("close без слушателей:", socket.emit("close"));
try {
  socket.emit("error", new Error("read ECONNRESET"));
  console.log("после error");
} catch (err) {
  console.log("вылетело из emit:", err.message);
}

// 5: двенадцать подписок на одно событие
const db = new Emitter();
for (let id = 1; id <= 12; id++) db.on("change", () => {});
console.log("слушателей change:", db.listenerCount("change"));
console.log("конец синхронного кода");

На этом классе стоит следующая статья: читающий и пишущий поток — это излучатель с буфером и событиями data и end. А ловушки отсюда вернутся в главе про HTTP: сервер, запрос и сокет в Node — тоже излучатели.