WebSocket

WebSocket — постоянное двустороннее соединение между браузером и сервером. В Winter WebSocket-сервер — это процесс: вы описываете, кого пускать и что делать с сообщением, а фреймворк держит соединения, проверяет, что клиенты живы, и мягко закрывает всё при остановке.

Что такое WebSocket и зачем

HTTP устроен как разговор, который всегда начинает клиент: браузер спросил, сервер ответил, обмен окончен. Пока сервер не спросили, сказать он ничего не может.

Проблема. Многим приложениям нужно и обратное направление: сервер сам должен сообщить что-то сейчас. Пришло сообщение в чат, изменился курс, заказ перешёл в статус «оплачен», другой игрок сделал ход. Обходной путь — опрашивать сервер раз в секунду («есть что-нибудь новое?»). Это запрос в секунду от каждого открытого окна, почти всегда с ответом «нет», и всё равно с задержкой до секунды.

Решение — WebSocket (протокол RFC 6455). Браузер отправляет обычный HTTP-запрос с просьбой «переключиться» (upgrade), сервер соглашается, и то же самое TCP-соединение остаётся открытым. Дальше обе стороны отправляют друг другу сообщения когда хотят, без новых запросов, пока соединение не закроют.

Что берёт на себя Winter. Держать тысячи открытых соединений в одном процессе, отвечать на служебные сигналы протокола, замечать клиентов, которые пропали без предупреждения, и аккуратно закрывать всё при остановке — однотипная и полная ловушек работа. Winter даёт стереотип WebSocket: вы пишете несколько хуков («кого пускать», «что делать с сообщением», «кто-то ушёл»), фреймворк делает остальное. WebSocket — это процесс: он запускается, останавливается и показывает состояние так же, как любой другой.

Быстрый старт

Наследуйте WebSocket и реализуйте единственный обязательный хук — onMessage():

<?php

namespace Main;

use Flytachi\Winter\Kernel\WebSocket\Connection;
use Flytachi\Winter\Kernel\WebSocket\Frame;
use Flytachi\Winter\Kernel\WebSocket\WebSocket;

final class ChatSocket extends WebSocket
{
  protected function onMessage(Connection $conn, Frame $frame): void
  {
      // каждое сообщение — всем, кроме отправителя
      $this->broadcast($frame->data, except: $conn);
  }
}

Объявите его в манифесте приложения и укажите порт:

#[EnableWeb]
#[EnableWebSocket(\Main\ChatSocket::class, port: 8001)]
final class Application extends WinterApplication { /* ... */ }

После php call run рядом с веб-сервером поднимется процесс сокета. Подключиться можно прямо из консоли браузера:

const ws = new WebSocket('ws://localhost:8001');
ws.onmessage = (e) => console.log('пришло:', e.data);
ws.onopen = () => ws.send('привет всем');

Откройте две вкладки: то, что отправлено из одной, появится в другой. Это уже рабочий чат.

Заготовку класса можно сгенерировать командой make:

php call make .Chat -W      # → main/ChatSocket.php

Запуск

Порт обязателен всегда: один процесс — один порт. Запустить сокет можно тремя способами.

Вместе с приложением — атрибутом на классе приложения. Это основной способ:

#[EnableWeb]
#[EnableWebSocket(ChatSocket::class, port: 8001)]
#[EnableWebSocket(NotifySocket::class, port: 8002, host: '127.0.0.1')]
final class Application extends WinterApplication { /* ... */ }

Отдельно из консоли — как любой процесс, но с --port:

php call process main.ChatSocket start --port=8001          # на переднем плане
php call process main.ChatSocket start --port=8001 -d       # в фоне
php call process main.ChatSocket status                     # что с ним
php call process main.ChatSocket stop                       # мягко остановить

Из кода — в фоне, например из админки:

ChatSocket::dispatch(port: 8001);

Параметр host — сетевой интерфейс. По умолчанию 0.0.0.0 — все интерфейсы. 127.0.0.1 оставляет сокет доступным только с этой машины — так делают, когда перед ним стоит прокси на том же хосте.

Настройки

Как работает сокет, задают свойства класса:

final class ChatSocket extends WebSocket
{
  protected ?string $path = '/chat';
  protected ?array $allowedOrigins = ['https://example.com'];
  protected float $heartbeat = 30.0;
  protected int $maxConnections = 10_000;
  protected int $maxFrameSize = 64 * 1024;
  protected float $grace = 10.0;

  // ...
}

Зависимости внедряются обычным #[Autowired] — сокет создаёт контейнер, как и любой процесс.

Хуки

Хуки — это методы, которые вы переопределяете. Обязателен только onMessage().

onStart

Подготовка и фоновая работа процесса — до приёма первого соединения.

Вызывается один раз, когда сервер уже создан, но ещё не принимает клиентов. Фреймворк дожидается окончания хука, поэтому всё, что здесь загружено, гарантированно готово к первому подключению. Здесь же запускают работу, не привязанную к конкретному клиенту: подписку на канал Redis, периодическую рассылку, чтение внешнего потока.

Синтаксис

protected function onStart(): void

Параметры

Нет.

Ошибки

Исключение из onStart() останавливает процесс: сокет не стартует, ошибка попадает в лог. Под call run вместе с веб-сервером мастер поднимет процесс заново.

Пример

protected function onStart(): void
{
  $this->rooms = $this->roomRepository->findAllActive();   // прогрев

  $this->spawn(function (): void {
      while ($this->isRunning()) {
          $this->broadcast(json_encode(['online' => $this->count()]));
          $this->sleep(5);
      }
  });
}

onHandshake

Решает, пускать ли клиента.

Вызывается до переключения протокола: в этот момент соединение ещё обычный HTTP-запрос. Здесь проверяют доступ — токен, cookie сессии, заголовок Origin — и сохраняют в соединении то, что понадобится дальше (например, найденного пользователя), чтобы не искать его заново на каждом сообщении. Отправлять сообщения отсюда нельзя: WebSocket ещё не открыт.

Синтаксис

protected function onHandshake(Connection $conn, HttpRequest $request): bool

Параметры

$conn — будущее соединение. Методы set(), get(), has() работают, и сохранённое переходит в открытое соединение. send() бросает LogicException.

$request — запрос, с которым клиент пришёл: заголовки, cookie, параметры строки запроса, путь, IP. Тот же объект, что получают контроллеры.

Возвращает

true — пустить: соединение переключается на WebSocket, затем вызывается onOpen(). false — отказать: клиент получает ответ 403.

По умолчанию хук возвращает true, то есть пускает всех.

Ошибки

Чтобы отказать с точным статусом, бросьте ResponseException — клиент получит её код и сообщение, так же, как из контроллера (работают и обработчики #[AdviceException]). Любое другое исключение — это ответ 500 и вызов onError().

Пример

use Flytachi\Winter\Base\HttpCode;
use Flytachi\Winter\Kernel\Http\Response\ResponseException;

protected function onHandshake(Connection $conn, HttpRequest $request): bool
{
  $token = $request->getQueryParams()['token'] ?? null;
  $user = $token === null ? null : $this->tokens->verify($token);

  if ($user === null) {
      return false;                                                  // 403
  }
  if ($user->isBlocked()) {
      throw new ResponseException('account blocked', HttpCode::FORBIDDEN);
  }

  $conn->set('user', $user);
  return true;
}

protected ?array $allowedOrigins = ['https://example.com', 'https://app.example.com'];

onOpen

Соединение открыто — отправлять уже можно.

Вызывается сразу после переключения протокола. Место для приветствия, отправки истории сообщений, оповещения остальных о новом участнике.

Синтаксис

protected function onOpen(Connection $conn): void

Параметры

$conn — открытое соединение, со всем, что сохранено в onHandshake().

Пример

protected function onOpen(Connection $conn): void
{
  $name = $conn->get('user')->name;

  $conn->send(json_encode(['type' => 'welcome', 'name' => $name]));
  $this->broadcast(json_encode(['type' => 'joined', 'name' => $name]), except: $conn);
}

onMessage

Пришло сообщение от клиента. Единственный обязательный хук.

Вызывается на каждое текстовое или бинарное сообщение. Сообщения одного соединения приходят строго по очереди: следующее не будет обработано, пока не закончен этот вызов. Сообщения разных соединений обрабатываются одновременно — медленный хук одного клиента не задерживает остальных.

Синтаксис

abstract protected function onMessage(Connection $conn, Frame $frame): void

Параметры

$conn — соединение, от которого пришло сообщение.

$frame — само сообщение: $frame->data — содержимое строкой, $frame->isBinary — true для бинарного сообщения.

Служебные сигналы протокола (ping, pong, закрытие) сюда не попадают — на них отвечает фреймворк.

Пример

Формат сообщений выбираете вы. Распространённый приём — JSON с полем type, по которому одно соединение несёт разные виды сообщений:

protected function onMessage(Connection $conn, Frame $frame): void
{
  $msg = json_decode($frame->data, true);

  match ($msg['type'] ?? null) {
      'join' => $conn->set('room', (int) $msg['room']),
      'say'  => $this->sendToRoom($conn->get('room'), $conn->get('user')->name, $msg['text']),
      default => $conn->send(json_encode(['type' => 'error', 'text' => 'unknown type'])),
  };
}

private function sendToRoom(?int $room, string $from, string $text): void
{
  $payload = json_encode(['type' => 'say', 'from' => $from, 'text' => $text]);

  foreach ($this->connections() as $member) {
      if ($member->get('room') === $room) {
          $member->send($payload);
      }
  }
}

onClose

Соединение закрылось.

Вызывается, кто бы ни закрыл соединение и по какой бы причине: клиент ушёл, сервер вызвал close(), клиент перестал отвечать, процесс останавливается. Только для соединений, которые прошли onHandshake() — отказанный клиент сюда не попадает.

Синтаксис

protected function onClose(Connection $conn, int $code, string $reason): void

Параметры

$code — код закрытия:

$reason — текст причины, если он был.

Пример

protected function onClose(Connection $conn, int $code, string $reason): void
{
  $user = $conn->get('user');
  if ($user !== null) {
      $this->broadcast(json_encode(['type' => 'left', 'name' => $user->name]));
  }
}

onError

Исключение в одном из хуков.

Вызывается, когда onOpen(), onMessage() или onClose() бросили исключение, а также при исключении в onHandshake(), которое не является ResponseException. По умолчанию записывает ошибку в лог — с номером соединения и адресом клиента — и оставляет соединение открытым: одно неудачное сообщение не обрывает всю сессию, так же как ошибка в контроллере не закрывает keep-alive соединение.

Синтаксис

protected function onError(Connection $conn, \Throwable $e): void

Параметры

$conn — соединение, на котором произошла ошибка.

$e — исключение.

Пример

Переопределите, если клиенту нужно сообщить об ошибке или соединение лучше закрыть:

protected function onError(Connection $conn, \Throwable $e): void
{
  parent::onError($conn, $e);   // в лог

  $conn->send(json_encode(['type' => 'error', 'text' => 'something went wrong']));

  $errors = (int) $conn->get('errors') + 1;
  $conn->set('errors', $errors);
  if ($errors >= 5) {
      $conn->close(1008, 'too many errors');
  }
}

Унаследованные от Process

WebSocket — процесс, поэтому доступны и его хуки: onTerminate(), onReload(), onUser1(), onUser2() — реакция на сигналы, и onShutdown() — последнее действие перед выходом. Закрывать соединения при остановке не нужно — это делает фреймворк. Подробнее — в Процессах.

Методы сокета

Эти методы вызываются на $this внутри класса.

broadcast

Отправляет текстовое сообщение всем открытым соединениям этого процесса.

Сообщение ставится в очередь каждого соединения отдельно, поэтому медленный клиент не задерживает доставку остальным, а сам метод возвращается сразу.

Синтаксис

final protected function broadcast(string $data, ?Connection $except = null): int

Параметры

$data — текст сообщения.

$except — соединение, которое пропустить; обычно отправитель. По умолчанию null — всем.

Возвращает

Число соединений, которым сообщение поставлено в очередь.

Пример

$this->broadcast(json_encode(['type' => 'say', 'text' => $frame->data]), except: $conn);

connections

Возвращает открытые соединения процесса.

Нужен для выборочной отправки: в комнату, конкретному пользователю, по роли.

Синтаксис

final protected function connections(): array

Возвращает

Массив Connection, ключ — номер соединения.

Пример

foreach ($this->connections() as $conn) {
  if ($conn->get('user')?->id === $userId) {
      $conn->send($text);
  }
}

connection

Находит открытое соединение по номеру.

Синтаксис

final protected function connection(int $id): ?Connection

Параметры

$id — номер соединения ($conn->id).

Возвращает

Соединение или null, если оно уже закрыто.

count

Возвращает число открытых соединений.

Синтаксис

final protected function count(): int

Пример

$this->logger->info('online: ' . $this->count());

Из Process

Доступно всё, что есть у процесса: spawn() — фоновая задача, sleep() — пауза, которая прерывается при остановке, isRunning() — false, когда процесс останавливается, $this->logger — логгер.

Соединение — Connection

Connection — это объект одного клиентского соединения. Один и тот же объект проходит через все хуки соединения, от onHandshake() до onClose(), поэтому то, что положено в него методом set(), доступно при следующем сообщении и исчезает вместе с соединением.

send

Отправляет текстовое сообщение.

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

Синтаксис

public function send(string $data): bool

Параметры

$data — текст сообщения.

Возвращает

true — сообщение поставлено в очередь. false — соединение уже закрыто или закрывается, либо клиент перестал читать и его очередь (1024 сообщения) заполнена; в этом случае сообщение отбрасывается.

Ошибки

LogicException — если вызвать в onHandshake(), до переключения протокола.

sendBinary

Отправляет бинарное сообщение. В остальном — как send().

Синтаксис

public function sendBinary(string $data): bool

close

Закрывает соединение.

Сообщения, отправленные до вызова, ещё уйдут; затем клиенту уходит сигнал закрытия с кодом и причиной, и вызывается onClose() с этим же кодом. Повторный вызов ничего не делает.

Синтаксис

public function close(int $code = 1000, string $reason = ''): void

Параметры

$code — код закрытия. 1000 — обычное, 1008 — нарушение правил, 1011 — ошибка сервера, 4000–4999 — ваши собственные.

$reason — короткий текст для клиента (не больше 123 байт).

Пример

$conn->send(json_encode(['type' => 'kicked']));
$conn->close(4001, 'kicked by moderator');

isOpen

Возвращает true, пока в соединение можно отправлять. false в onHandshake() и после закрытия.

Синтаксис

public function isOpen(): bool

request

Возвращает запрос, с которым клиент подключился: заголовки, cookie, параметры строки запроса. Доступен всю жизнь соединения.

Синтаксис

public function request(): HttpRequest

set, get, has

Хранилище значений соединения.

Синтаксис

public function set(string $key, mixed $value): void
public function get(string $key): mixed          // null, если ключа нет
public function has(string $key): bool           // сохранённый null тоже считается

Пример

$conn->set('room', 42);

if ($conn->has('room')) {
  $room = $conn->get('room');
}

Сообщение — Frame

Frame — входящее сообщение клиента, то, что получает onMessage().

Жизненный цикл соединения

подключение
├─ путь не совпал с $path              → 404
├─ нет заголовка Upgrade: websocket   → 426
├─ Origin не из $allowedOrigins       → 403
├─ достигнут $maxConnections           → 503
└─ onHandshake()
      ├─ false                          → 403
      ├─ ResponseException              → её статус
      ├─ другое исключение              → 500 + onError()
      └─ true → переключение протокола → onOpen()
                  │
                  ├─→ onMessage()  (исключение → onError(), соединение живёт)
                  │      ↑___|
                  │
                  └─ клиент закрыл / нет ответа на ping / close() /
                     слишком большое сообщение / остановка процесса
                                        → onClose($code, $reason)

Проверка живости (heartbeat)

Клиент может исчезнуть, не закрыв соединение: телефон уехал в метро, ноутбук закрыли, роутер выбросил соединение из таблицы. Сервер об этом не узнает — для него соединение открыто, сообщения «успешно» уходят в никуда, а место в $maxConnections занято призраком. Кроме того, прокси обрывают молчащие соединения: у nginx по умолчанию через 60 секунд тишины.

Поэтому фреймворк раз в $heartbeat секунд отправляет клиенту служебный сигнал ping. Браузер отвечает на него pong сам — в JavaScript ничего писать не нужно. Если до следующей проверки от клиента ничего не пришло, соединение закрывается с кодом 1001, и вызывается onClose(). Любое сообщение от клиента тоже считается признаком жизни.

Значение по умолчанию — 30 секунд, меньше таймаута nginx. $heartbeat = 0 выключает проверку — например, для внутренних сервисов без прокси, где клиент проверяет связь сам.

Остановка

По stop, SIGTERM или остановке всего приложения сокет:

1. перестаёт принимать новые подключения;
2. отправляет каждому открытому соединению закрытие с кодом 1001 («going away») — браузер получает onclose с этим кодом и может переподключиться к другому экземпляру;
3. ждёт хуки, которые ещё выполняются, — не дольше $grace секунд;
4. вызывает onShutdown() и завершается.

За прокси

В продакшене сокет обычно стоит за nginx на том же домене, что и сайт. Тогда фронту не нужно знать про порт: браузер подключается к wss://example.com/ws/chat, а прокси передаёт соединение на порт процесса и заодно берёт на себя TLS.

location /ws/chat {
  proxy_pass http://app:8001;
  proxy_http_version 1.1;
  proxy_set_header Upgrade $http_upgrade;
  proxy_set_header Connection "upgrade";
  proxy_read_timeout 120s;
}

location /ws/notify {
  proxy_pass http://app:8002;
  proxy_http_version 1.1;
  proxy_set_header Upgrade $http_upgrade;
  proxy_set_header Connection "upgrade";
}

Заголовки Upgrade и Connection обязательны: без них nginx не передаёт просьбу о переключении, и сокет ответит 426.

Отправка из контроллера

Частая задача: «заказ оплачен — сообщить пользователю в открытой вкладке». Контроллер работает в веб-воркере, а соединения живут в процессе сокета — это разные процессы, и напрямую вызвать send() из контроллера нельзя. Нужен посредник; удобнее всего — канал Redis (pub/sub): контроллер публикует, сокет слушает в onStart() и раздаёт.

use Flytachi\Winter\Redis\RedisPool;

public function markPaid(int $id): array
{
  $order = $this->orders->markPaid($id);

  RedisPool::store(MainRedisConfig::class)->publish('notify', json_encode([
      'userId' => $order->userId,
      'text'   => "Заказ №{$order->id} оплачен",
  ]));

  return ['status' => 'ok'];
}

use Flytachi\Winter\Redis\RedisPool;

protected float $grace = 5.0;   // подписка сама не завершается — не ждать её вечно

protected function onStart(): void
{
  $this->spawn(function (): void {
      // подписка занимает соединение целиком — берём своё, мимо пула
      $config = RedisPool::dedicated(MainRedisConfig::class);
      $redis  = $config->connection();
      $redis->setOption(\Redis::OPT_READ_TIMEOUT, -1);

      $redis->subscribe(['notify'], function (\Redis $redis, string $channel, string $payload): void {
          $msg = json_decode($payload, true);

          foreach ($this->connections() as $conn) {
              if ($conn->get('user')?->id === $msg['userId']) {
                  $conn->send($msg['text']);
              }
          }
      });
  });
}

Тот же приём решает задачу нескольких экземпляров приложения: каждый экземпляр сокета подписан на канал и раздаёт сообщение своим клиентам. Ядро не навязывает способ доставки — можно использовать брокер сообщений или что-то своё.

Пример: чат с комнатами

Самодостаточный пример: вход по токену, комнаты, сообщения в комнату, уход.

<?php

namespace Main;

use Flytachi\Winter\DI\Attribute\Autowired;
use Flytachi\Winter\Kernel\Http\Contracts\HttpRequest;
use Flytachi\Winter\Kernel\WebSocket\Connection;
use Flytachi\Winter\Kernel\WebSocket\Frame;
use Flytachi\Winter\Kernel\WebSocket\WebSocket;

final class RoomSocket extends WebSocket
{
  protected int $maxFrameSize = 16 * 1024;

  #[Autowired]
  private TokenService $tokens;

  protected function onHandshake(Connection $conn, HttpRequest $request): bool
  {
      $user = $this->tokens->verify($request->getQueryParams()['token'] ?? '');
      if ($user === null) {
          return false;
      }
      $conn->set('name', $user->name);
      return true;
  }

  protected function onMessage(Connection $conn, Frame $frame): void
  {
      $msg = json_decode($frame->data, true);

      if (($msg['type'] ?? null) === 'join') {
          $conn->set('room', (string) $msg['room']);
          $this->toRoom($conn->get('room'), ['type' => 'joined', 'name' => $conn->get('name')]);
          return;
      }

      if (($msg['type'] ?? null) === 'say' && $conn->has('room')) {
          $this->toRoom($conn->get('room'), [
              'type' => 'say',
              'name' => $conn->get('name'),
              'text' => mb_substr((string) $msg['text'], 0, 2000),
          ]);
      }
  }

  protected function onClose(Connection $conn, int $code, string $reason): void
  {
      if ($conn->has('room')) {
          $this->toRoom($conn->get('room'), ['type' => 'left', 'name' => $conn->get('name')]);
      }
  }

  private function toRoom(string $room, array $message): void
  {
      $payload = json_encode($message);
      foreach ($this->connections() as $member) {
          if ($member->get('room') === $room) {
              $member->send($payload);
          }
      }
  }
}

const ws = new WebSocket('ws://localhost:8001?token=YOUR_TOKEN');
ws.onmessage = (e) => console.log(JSON.parse(e.data));
ws.onopen = () => {
  ws.send(JSON.stringify({ type: 'join', room: 'general' }));
  ws.send(JSON.stringify({ type: 'say', text: 'всем привет' }));
};

Дальше

• Процессы — основа, на которой построен сокет: сигналы, spawn(), управление
• Состав приложения — где объявляется #[EnableWebSocket]
• Команда process — запуск, состояние и остановка из консоли
• Команда make — заготовка сокета флагом -W
• Redis — каналы для связи сокета с контроллерами и другими экземплярами