События в реальном времени и WebSocket#

Используйте server-sent events для односторонних обновлений и WebSocket для постоянного двустороннего соединения. Elang включает долговременную историю событий в базе данных, разрешения доступа к закрытым каналам, курсоры воспроизведения и среду выполнения WebSocket на основе Workerman. Стартовый проект API демонстрирует оба транспорта на заметках, принадлежащих пользователям.

Установка и запуск#

composer create-project --repository=https://elangphp.com/packages elang/api my-api 0.1.0-dev
cd my-api
composer require elang/realtime:0.1.0-dev
php elang serve

Откройте второй терминал в каталоге приложения:

php elang realtime:serve --port=8081 --workers=1

Для WebSocket требуется CLI PHP в Linux или macOS с pcntl и posix. Эти функции управления процессами должны быть включены в отдельной конфигурации CLI PHP; некоторые панели хостинга отключают их по умолчанию. Сохраняйте ограничения веб-пула отдельно. В рабочей среде также нужны диспетчер процессов и обратный прокси с TLS для wss://. SSE работает через обычное HTTP-приложение без необязательного пакета WebSocket.

Публикация после защищённой операции#

use Elang\Realtime\Events;

$events->publish('user:'.$user->id.':notes', 'note.created', [
    'note' => ['id' => 42, 'title' => 'A shared update'],
]);

Определяйте канал на основании аутентифицированного пользователя и серверной авторизации. Не принимайте из браузера произвольный канал без ограничений. Записывайте данные предметной области и соответствующее событие в одной транзакции Database, чтобы отменённые изменения не становились видимыми событиями. Данные передаются в JSON и ограничены 16 КБ.

Подписка через SSE#

Стартовый проект API предоставляет GET /api/notes/events с bearer-токеном notes:read. При повторном подключении передавайте Last-Event-ID. Потоки закрываются через 20 секунд, отправляют сигналы поддержания соединения и воспроизводят сохранённые события по порядку. Встроенный EventSource не позволяет задать заголовок Authorization; вместо него используйте включённый клиент на основе fetch.

const controller = new AbortController();
await api.events('/api/notes/events', {
  signal: controller.signal,
  onEvent: ({ id, event, data }) => console.log(id, event, data),
});

api создаётся через createClient из файла public/assets/elang-client.js стартового проекта API. См. интеграцию с React и Next.js. Остановите поток с помощью controller.abort(). Вспомогательный клиент повторно подключается после обычного завершения потока по тайм-ауту; сетевые ошибки и повторный вход обрабатывайте в приложении. Эндпоинт повторно проверяет bearer-токен, пока соединение открыто. Новая подписка без курсора начинается с нуля и воспроизводит сохранённую историю.

Подключение через WebSocket#

Выполните POST /api/realtime-token с bearer-токеном notes:read. Ответ содержит зашифрованное разрешение со сроком действия пять минут для канала заметок этого пользователя. Токен notes:write дополнительно разрешает команду note.create. Разрешение передаётся в подпротоколе WebSocket, а не в URL.

const socket = await api.socket({ url: 'ws://127.0.0.1:8081', after: 0 });
socket.onmessage = message => console.log(JSON.parse(message.data));
socket.onopen = () => {
  socket.send(JSON.stringify({ type: 'ping' }));
  // With a write-scoped capability:
  socket.send(JSON.stringify({
    type: 'note.create',
    payload: { title: 'From a socket', body: 'Stored by Elang' },
  }));
};

Отправляйте команды после открытия соединения. Для локальной разработки подключайтесь непосредственно к ws://127.0.0.1:8081/realtime/socket, используя возвращённый токен и протоколы ['elang', 'token.' + token]. Метод socket() клиента fetch использует источник API, если url не указан; в рабочей среде направляйте /realtime/socket через обратный прокси. Для отдельного локального порта WebSocket используйте url, как показано в примере. При установлении соединения разрешаются источник, настроенный в APP_URL, и явно заданные записи realtime.origins. Стартовый проект API читает последние из FRONTEND_URL. Проверки CORS и источника WebSocket выполняются отдельно; права на каналы и команды по-прежнему определяются зашифрованным разрешением.

Новые соединения без after начинаются с последнего события. Сохраняйте последний полученный ID и запрашивайте новое разрешение при повторном подключении. Срок действия разрешения истекает в пределах пяти минут. Смена пароля отзывает его через версию аутентификации пользователя; соединение проверяет действительность каждые пять секунд и перед командами. Отзыв исходного API-токена не отзывает уже выданное разрешение WebSocket; короткий срок действия ограничивает время его использования.

Настройка обратного прокси в рабочей среде#

Направьте путь WebSocket на локальный демон, работающий под управлением диспетчера процессов, в конфигурации виртуального хоста HTTPS. Для Nginx:

location /realtime/socket {
    proxy_pass http://127.0.0.1:8081;
    proxy_http_version 1.1;
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection "upgrade";
    proxy_set_header Host $host;
    proxy_read_timeout 75s;
    proxy_buffering off;
}

Проверьте конфигурацию виртуального хоста перед перезагрузкой Nginx. Оставьте демон привязанным к loopback-адресу и используйте wss:// из интерфейса, работающего по HTTPS. Пока соединение открыто, отправляйте сигнал поддержания соединения на уровне приложения:

const heartbeat = setInterval(() => {
  if (socket.readyState === WebSocket.OPEN) {
    socket.send(JSON.stringify({ type: 'ping' }));
  }
}, 20000);
socket.onclose = () => clearInterval(heartbeat);

После разрыва соединения подключайтесь повторно с новым разрешением и последним полученным идентификатором события. Сервер отвечает на JSON-сообщение ping сообщением pong; сигнал поддержания соединения сохраняет активным проксируемое соединение, которое иначе простаивало бы.

Определение команд приложения#

use Elang\Realtime\Messages;
use Elang\Auth\User;

$app->container->get(Messages::class)->register(
    'chat.send',
    function (array $payload, User $user, string $channel): array {
        // Validate payload, authorize the resource, persist, then publish.
        return ['accepted' => true];
    }
);

Выполняться могут только зарегистрированные команды, явно перечисленные при выдаче разрешения SocketTokens::issue. Сама регистрация не предоставляет клиенту никаких прав. Приведённый выше обработчик показывает только сигнатуру; реализуйте валидацию и сохранение данных, прежде чем использовать его для чата.

Эксплуатация и измерение#

Среда выполнения WebSocket ограничивает размер сообщений до 32 КБ, частоту до 60 клиентских сообщений в минуту, буфер отправки до 256 КБ и число соединений до 1000 на процесс. Медленные клиенты отключаются. Это защитные ограничения, а не гарантии измеренной пропускной способности. Текущий механизм событий опрашивает общую базу данных SQL каждые 500 мс для каждого соединения. Измеряйте нагрузку на базу данных и число соединений перед масштабированием; адаптер Redis или специализированного брокера остаётся в планах. Одно соединение SSE занимает HTTP-процесс в течение своего короткого потока, поэтому выделяйте отдельный пул процессов для потокового трафика.

Сохраняйте не менее 24 часов истории воспроизведения, запуская php elang realtime:prune по расписанию. Более старые курсоры воспроизводят только сохранившиеся события. После долгого отсутствия клиенты должны обновлять снимок актуального состояния ресурса. История событий не является бухгалтерским регистром или постоянным журналом аудита.