WebSocket ​

В чате поддержки пишут оба. Пользователь задаёт вопрос, оператор отвечает, и никто никого не ждёт: сообщения идут в обе стороны тогда, когда их написали.

Можно ли собрать такой чат из уже знакомых инструментов? Вниз, от сервера к браузеру, поедет SSE, это мы умеем со вчерашней главы. А вверх? А вверх придётся на каждую реплику пользователя отправлять отдельный POST. Новый запрос, заголовки, ответ, разрыв. И так на каждое «спасибо, помогло». Работать будет, но не особо.

Для разговора в обе стороны существует отдельный протокол, WebSocket.

Устроен он неожиданно скромно. Клиент отправляет обычный HTTP-запрос с заголовком Upgrade: предложением перейти на другой протокол. Сервер соглашается: 101 Switching Protocols. Всё, HTTP закончился. По тому же TCP-соединению теперь ходят сообщения, в обе стороны, без запросов и без ответов.

use TrueAsync\WebSocket;

$server->addWebSocketHandler(function (WebSocket $ws) {
    foreach ($ws as $msg) {
        $ws->send('echo: ' . $msg->data);
    }
});

Рукопожатие сервер берёт на себя целиком: обработчик вызывается, когда соединение уже переключено, и получает готовый объект. Модель знакомая: одно соединение — одна корутина. WebSocket реализует Iterator, и именно итератор усыпляет корутину до следующего сообщения. Клиент ушёл — цикл кончился — сервер сам закрыл соединение кодом 1000 Normal. А обычные HTTP-обработчики продолжают работать рядом, на том же порту.

Чат-комната ​

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

use TrueAsync\HttpRequest;

/** @var SplObjectStorage<WebSocket, string> $room */
$room = new SplObjectStorage();

$server->addWebSocketHandler(function (WebSocket $ws, HttpRequest $req) use ($room) {
    $name = $req->getQueryParam('name', 'guest');
    $room->attach($ws, $name);
    $ws->send("Добро пожаловать, в комнате: {$room->count()}");

    try {
        foreach ($ws as $msg) {
            foreach ($room as $peer) {
                if ($peer !== $ws) {
                    $peer->trySend("$name: {$msg->data}");
                }
            }
        }
    } finally {
        $room->detach($ws);
    }
});

Тридцать строчек, а внутри сошлись сразу три нити из первой серии. Не пробегайте мимо, каждая заслуживает паузы.

Первая: комната — это просто объект. Обычный SplObjectStorage, общий для всех корутин, без единой блокировки. Глава про каналы обещала, что в одном потоке так можно, и вот оно в деле.

Вторая: finally. Участник может выйти культурно, может пропасть вместе с Wi-Fi, а корутину может отменить сам сервер при остановке. Три сценария, один finally, и призрак гарантированно не останется в комнате. Кооперативная отмена, как она есть.

Третья: писать в чужое соединение можно из любой корутины. send() и trySend() для этого безопасны, сервер сам следит, чтобы кадры разных отправителей не перемешались на проводе. А вот читать — только из одной. Второй одновременный recv() на том же соединении получит исключение, и это справедливо: у байтового потока не бывает двух читателей с осмысленной семантикой.

Медленный клиент не остановит комнату ​

Вы заметили, что в рассылке стоит trySend, а не send? Это не опечатка, это решение, и его стоит проговорить.

send() при заполненном буфере делает что? Правильно, обратное давление: усыпляет корутину, пока клиент не разгребёт накопленное. Для личного ответа — то, что надо. А теперь представьте это в цикле рассылки. Один участник заехал в тоннель со своим мобильным интернетом, его буфер полон, и... вся комната ждёт. Сто человек не получают сообщений, потому что один плохо ловит.

trySend() не ждёт никогда. Либо сообщение принято в буфер и true, либо буфер полон и false, сообщение выброшено. Чат осознанно жертвует сообщениями отстающего, чтобы не наказывать всех. Жестоко? Для чата — нет: пропущенная реплика в комнате не трагедия. Если в вашей задаче терять нельзя, берите send() и миритесь с паузами или стройте очередь на каждого клиента. Обе стратегии честные, выбор за вами.

На крайний случай есть и предохранитель: если send() провисел в ожидании дольше таймаута записи, он бросит WebSocketBackpressureException. Это про клиента, который держит соединение открытым, но не читает вовсе.

Комнаты без бухгалтерии ​

SplObjectStorage своё дело сделал, но посчитайте, во что он нам обошёлся: общий объект, цикл рассылки и finally, чтобы выметать призраков. И всё ради того, чтобы реплика, набранная одним, дошла до остальных. Желание настолько частое, что сервер исполняет его напрямую. Соединение подписывается на имя; сообщение, опубликованное на это имя, доходит до всех подписанных. Тот же чат, только без бухгалтерии:

$server->addWebSocketHandler(function (WebSocket $ws, HttpRequest $req) {
    $room = $req->getQueryParam('room', 'lobby');
    $name = $req->getQueryParam('name', 'guest');

    $ws->subscribe("chat/$room");

    foreach ($ws as $msg) {
        $ws->publish("chat/$room", "$name: {$msg->data}");
    }
});

Хранилище исчезло, а вместе с ним цикл и finally. subscribe() помещает это соединение в комнату, publish() рассылает реплику всем, кто в ней, а закрывшееся соединение само покидает свои комнаты. publish() держит ту же сделку, что мы только что заключили с trySend(): у кого буфер забит, тот теряет реплику, а не тормозит комнату, и отправителя это никогда не блокирует.

Имя — это не просто ярлык, это фильтр в стиле MQTT. Уровни разделяются через /, + совпадает с одним уровнем, а # — с остатком. Подпишитесь на chat/+/typing — и услышите сигнал набора текста сразу из всех комнат; subscriberCount("chat/general") подскажет, сколько человек слушают. И ещё одно, тихо важное, чем мы воспользуемся в следующей главе: тема не привязана ни к одному соединению и даже ни к одному массиву в памяти. Запомните это.

Кто пустил его в чат? ​

Сейчас в комнату заходит кто угодно. Хотите проверять на входе — объявите у обработчика третий параметр, и сервер передаст объект рукопожатия:

use TrueAsync\WebSocketUpgrade;

$server->addWebSocketHandler(
    function (WebSocket $ws, HttpRequest $req, WebSocketUpgrade $upgrade) use ($room) {
        if (authenticate($req) === null) {
            $upgrade->reject(401, 'auth required');
            return;
        }

        // ... комната ...
    }
);

reject() отвечает обычной HTTP-ошибкой вместо переключения протокола. Через тот же объект согласуется и subprotocol, если клиент их предлагает.

Чат готов: комната в памяти, рассылка мгновенная, отстающие никого не тормозят. Запомните формулу, на которой всё это держится: «общее состояние в памяти процесса». Хорошенько запомните. В следующей главе мы включим несколько воркеров, чтобы занять все ядра машины, и эта безобидная формула взорвётся первой.