Уведомления через Server-Sent Events

Оглавление серии

Для отправки и получения реплик «Линия» использует двусторонний WebSocket. Но некоторые события нужны только от сервера к браузеру: например, уведомление, что обращение обновилось. Рассмотрим Server-Sent Events, сокращённо SSE. Этот транспорт оставляет HTTP-ответ открытым и передаёт текстовые события последовательно, а браузер получает их через EventSource.

Тринадцатое состояние stages/13 добавляет отдельный /events, не заменяя канал переписки. Уведомление содержит ссылочные идентификаторы, а сообщение остаётся в журнале комнаты. Пример имеет статус not_run: соединения, повторные подключения и ответы не выполнялись. Граница восстановления SSE определяется отдельным окном ста уведомлений в памяти процесса, поэтому оно не является долговечной очередью.

Открытый HTTP-ответ

Сервер устанавливает Content-Type: text/event-stream, запрещает кеширование и передаёт заголовки через flushHeaders(). После этого ответ не завершается обычным JSON. Каждое событие состоит из текстовых строк, а пустая строка отделяет одно событие от следующего. Если забыть завершение двумя переводами строки, браузер может продолжать ждать конец события и ничего не показать, хотя байты уже переданы.

Полный файл events.js показывает запись события, хранение короткого журнала и очистку подписчиков. Формат сериализуется из нашего объекта уведомления, а не из произвольного текста пользователя.

import { canEnter } from './identity.js';

const listeners = new Set();
const journal = [];
let seq = 0;

function writeEvent(response, event) {
  const frame = `id: ${event.id}\nevent: notification\n` +
    `data: ${JSON.stringify(event)}\n\n`;
  if (!response.write(frame)) response.end();
}

export function notify(message) {
  const event = {
    id: ++seq,
    roomId: message.roomId,
    messageId: message.messageId
  };
  journal.push(event);
  if (journal.length > 100) journal.shift();
  for (const listener of listeners) {
    if (canEnter(listener.actor, event.roomId)) {
      writeEvent(listener.response, event);
    }
  }
}

export function subscribe(request, response, actor) {
  response.set({
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
    'Connection': 'keep-alive'
  });
  response.flushHeaders();
  const raw = request.get('Last-Event-ID');
  const after = raw ? Number(raw) : seq;
  const first = journal[0]?.id || seq + 1;
  if (!Number.isSafeInteger(after) || after < first - 1 || after > seq) {
    response.write('event: resync\ndata: {}\n\n');
  } else {
    for (const event of journal) {
      if (event.id > after && canEnter(actor, event.roomId)) {
        writeEvent(response, event);
      }
    }
  }
  const listener = { response, actor };
  listeners.add(listener);
  const timer = setInterval(() => {
    if (!response.write(': keepalive\n\n')) response.end();
  }, 15000);
  response.on('close', () => {
    clearInterval(timer);
    listeners.delete(listener);
  });
}

export function closeEvents() {
  for (const listener of listeners) listener.response.end();
}

Строка event: notification задаёт имя события, на которое клиент подписывается отдельно. Строка data: содержит JSON, а id: сообщает курсор потока. Это текстовый формат SSE, отличный от формата обычного HTTP-ответа. Не отправляйте JSON без строковых префиксов в открытый поток и не воспринимайте каждый вызов write() как отдельный сетевой пакет: границы события определяются разделителем, а не способом записи байтов.

Ожидаемый кадр уведомления имеет такую структуру. Идентификаторы условные, поскольку пример не запускался.

id: 1
event: notification
data: {"id":1,"roomId":"ticket-17","messageId":"message-id"}

Уведомление и содержание различаются

После принятия сообщения обработчик вызывает notify(message). Событие передаёт комнату и идентификатор сообщения, но не повторяет текст переписки. В интерфейсе появляется надпись о наличии обновления. Такой контракт позволяет отдельному списку обращений узнать, что состояние стоит перечитать, не превращая каждый транспорт в ещё одну независимую копию полного журнала. Саму реплику текущая комната продолжает получать через WebSocket.

Счётчик SSE общий для потока уведомлений, а seq сообщения относится к комнате. Их нельзя подставлять друг вместо друга. Уведомление номер 20 может ссылаться на сообщение номер 3 второй комнаты, если между ними обновлялись другие обращения. Восстановление SSE отвечает за последовательность уведомлений, восстановление чата — за последовательность переписки. Совпадающее числовое значение не делает курсоры взаимозаменяемыми.

Каждый подписчик хранит серверный объект участника, а перед записью уведомления вызывается canEnter(). Поэтому Алиса не получает уведомления комнаты Бориса. Проверка выполняется при рассылке и при повторной выдаче сохранённого окна. Даже минимальное уведомление может раскрыть факт существования чужого обращения и время активности, поэтому исключать права только потому, что текст сообщения не передаётся, неправильно.

Повторное подключение EventSource

Браузерный EventSource умеет автоматически переподключаться и передавать последний полученный ID в Last-Event-ID. Сервер читает заголовок и выдаёт последующие записи, если курсор находится в сохранённом окне. Здесь не нужно добавлять секрет в query URL. Cookie учебной сессии используется обычным запросом на том же источнике, а сервер проверяет её до открытия потока. Это сохраняет общий способ идентичности приложения.

При слишком старом, неверном или находящемся впереди курсоре сервер отправляет resync. Событие означает, что клиент должен перечитать состояние, а не считать себя восстановленным. Наш демонстрационный интерфейс показывает соответствующую надпись; полноценный список обращений должен после неё запросить актуальные данные. Автоматическое восстановление транспорта не гарантирует восстановление предметных событий, если журнал уже удалил их.

В этом снимке ID потока не включает epoch, поэтому рестарт с новым малым счётчиком обнаруживается только когда старый курсор находится впереди или вне окна. Если числа случайно снова совпадут, одного числового ID недостаточно для точного определения старой эпохи. Для production следует включить идентификатор эпохи либо использовать долговечный курсор общего журнала. Это ограничение отдельного учебного SSE-потока, явно отличное от протокола истории чата.

Строка комментария : keepalive периодически передаётся без предметного события. Она помогает сохранять активность соединения и не вызывает обработчик notification. Промежуточный прокси может иметь собственный таймаут или буферизовать ответ, поэтому факт работы локального потока не подтверждает поведение будущей инфраструктуры. При выпуске необходимо отдельно проверить настройки прокси и реальное время прохождения событий; здесь никаких измерений не проводили.

Закрытие и медленный получатель

Когда ответ закрывается, обработчик close удаляет подписчика и очищает его таймер. Иначе карта продолжала бы держать response и actor, а интервалы обращались бы к завершённым потокам. Само исчезновение вкладки не очищает произвольные пользовательские структуры автоматически. Каждый созданный ресурс — запись подписчика, таймер, ответ — должен иметь определённый путь освобождения, особенно для долго живущих соединений.

Возвращаемое значение response.write() сообщает, удалось ли продолжить запись без накопления сверх порога внутреннего буфера. Учебная реализация при false завершает поток; браузер сможет переподключиться и воспользоваться окном. Более сложный вариант ждёт drain и сохраняет ограниченную очередь. Главное — не бесконечно игнорировать сигнал и не превращать медленного получателя в причину роста памяти всего сервера.

Для передачи команд SSE не подходит: EventSource получает события, но не предоставляет симметричную отправку сообщения в этом же канале. Команды можно выполнять отдельным fetch, однако договор отправки, подтверждений и идемпотентности всё равно потребуется. Выбор транспорта следует из направления обмена и модели восстановления, а не из желания использовать одно название API для всех задач интерфейса.

Формат и Last-Event-ID описаны в HTML Standard: Server-sent events. После этого состояния уведомления имеют собственный поток, курсор и политику очистки. Следующим шагом обобщим ограничения скорости и памяти для WebSocket, где открытое соединение тоже не гарантирует, что получатель успевает обработать события.