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() при заповненому буфері робить що? Правильно, backpressure: присипляє корутину, поки клієнт не розгребе накопичене. Для особистої відповіді, це те, що треба. А тепер уявіть це в циклі розсилки. Один учасник заїхав у тунель зі своїм мобільним інтернетом, його буфер повний, і... уся кімната чекає. Сто людей не отримують повідомлень, бо один погано ловить.

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, якщо клієнт його пропонує.

Чат готовий: кімната в пам'яті, розсилка миттєва, відсталі нікого не гальмують. Запам'ятайте формулу, на якій усе це тримається: «спільний стан у пам'яті процесу». Добряче запам'ятайте. У наступному розділі ми ввімкнемо кілька воркерів, щоб зайняти всі ядра машини, і ця безневинна формула вибухне першою.