Server-Sent Events (SSE)

OxPHP передає клієнтам дані в реальному часі за протоколом Server-Sent Events із вбудованим зворотним тиском (backpressure). Установіть у PHP-скрипті Content-Type: text/event-stream і викличте oxphp_stream_flush(). Решту OxPHP зробить сам.

Як це працює

Потокова відповідь проходить передбачуваний життєвий цикл:

Установлення заголовків. PHP-скрипт установлює Content-Type: text/event-stream через header() і записує рядки у форматі SSE за допомогою echo.
Перший flush відкриває потік. Перший виклик oxphp_stream_flush() надсилає клієнтові HTTP-заголовки й переводить відповідь у потоковий режим. З’єднання з клієнтом лишається відкритим.
Кожен flush надсилає фрагмент. Кожен наступний виклик oxphp_stream_flush() скидає буферизований вивід як новий фрагмент (chunk) і одразу доставляє його клієнтові.
Зворотний тиск вмикається, коли буфер заповнюється. OxPHP тримає між PHP-воркером і клієнтом внутрішній буфер обсягом до 64 фрагментів. Коли буфер повний — бо повільний клієнт ще не забрав попередні фрагменти, — oxphp_stream_flush() блокується, доки не звільниться місце. Це запобігає необмеженому зростанню споживання пам’яті.
Прибирання після завершення або відключення. Коли PHP-скрипт завершується, OxPHP коректно закриває з’єднання. Якщо клієнт відключається посеред потоку, скрипт дізнається про це на першому flush після того, як OxPHP помітить закриття з’єднання. OxPHP установлює PHP-прапорець connection_aborted() і, якщо скрипт не викликав ignore_user_abort(true), зупиняє скрипт усередині цього flush: виклик не повертається — саме так PHP зупиняє скрипт під PHP-FPM, коли його вивід більше неможливо записати. Скриптові, який викликав ignore_user_abort(true), flush повертає керування, і він продовжує працювати, а завершує його цикл перевірка connection_aborted() (див. Виявлення відключень клієнта).

Note

Щоб пропускна здатність лишалася рівною, робіть корисне навантаження подій невеликим. Великі payload швидко заповнюють буфер на 64 фрагменти, і PHP блокується на кожному flush.

Приклади

Базовий SSE-потік

php

<?php
header('Content-Type: text/event-stream');
header('Cache-Control: no-cache');
header('Connection: keep-alive');

for ($i = 0; $i < 100; $i++) {
    $data = json_encode(['counter' => $i, 'time' => microtime(true)]);
    echo "id: {$i}\n";
    echo "event: tick\n";
    echo "data: {$data}\n\n";
    oxphp_stream_flush();

    sleep(1);

    // Send a comment heartbeat every 15 seconds to keep proxies from closing idle connections
    if ($i % 15 === 0) {
        echo ": heartbeat\n\n";
        oxphp_stream_flush();
    }
}

Перевірка стану потокової передачі

Функція oxphp_is_streaming() показує, чи поточний запит уже перебуває в потоковому режимі. Це зручно в middleware або спільних обробниках запитів:

php

<?php
if (!oxphp_is_streaming()) {
    header('Content-Type: text/event-stream');
    header('Cache-Control: no-cache');
}

echo "data: {\"status\": \"connected\"}\n\n";
oxphp_stream_flush();

Виявлення відключень клієнта

Потік, який уже надіслав заголовки, дізнається, що його клієнт пішов, лише коли робить flush. За замовчуванням саме на цьому flush скрипт і зупиняється: oxphp_stream_flush() (або flush()) не повертається, тож умова while (!connection_aborted()) більше ніколи не обчислюється, а всі блоки finally, усередині яких був flush, пропускаються. PHP-FPM так само зупиняє скрипт, чий вивід більше неможливо записати.

Виконується далі лише завершення запиту: колбеки register_shutdown_function(), у яких connection_aborted() повертає 1, і деструктори об’єктів, що зникають разом із запитом. На відміну від PHP-FPM, який відкидає все, що вони записують, OxPHP зупиняє колбек на першому виводі, який той надсилає, і пропускає колбеки, зареєстровані після нього. Деструктори все одно виконуються, і перший із них, що надсилає вивід, зупиняється так само, а деструктори, черга яких настає після нього, пропускаються. Вивід, що утримується в буфері виводу (відкритому через ob_start() або тому, який відкриває output_buffering у php.ini), надсилається лише тоді, коли цей буфер скидається. Тримайте таке прибирання без виводу або викличте ignore_user_abort(true), як показано нижче.

Щоб прибрати за собою в самому скрипті — у блоці finally або після циклу, — викличте ignore_user_abort(true) і дайте циклу завершитися через connection_aborted(). Тоді flush, який виявить, що клієнт пішов, повертає керування, connection_aborted() відтоді повертає 1, і цикл виходить через свою умову:

php

<?php
header('Content-Type: text/event-stream');
header('Cache-Control: no-cache');

// A stream outlives max_execution_time, which is 30 s unless the PHP configuration sets another value
set_time_limit(0);

// Keep running when the client leaves, so that the loop below sees it and ends
ignore_user_abort(true);

$db = new PDO(/* ... */);

try {
    while (!connection_aborted()) {
        echo "data: " . json_encode(['ts' => time()]) . "\n\n";
        oxphp_stream_flush();
        sleep(1);
    }
} finally {
    $db = null; // also runs when the loop ends on a disconnect
}

`ignore_user_abort(true)` не дає потоку зупинитися

З цим викликом перевірка connection_aborted() — єдине, що завершує цикл: скрипт працює й після відключення клієнта, усе, що він записує, відкидається, а connection_aborted() повідомляє про відключення. Це той самий контракт, що й у будь-якому іншому SAPI. Такому циклу потрібне власне обмеження: потік, який попросив пережити свого клієнта, не перевіряє connection_aborted() і не завершується сам, утримує свій воркер аж до зупинки сервера.

Виклик діє лише на той запит, який його робить. Проте значення, з яким запит починається, враховується так само: ignore_user_abort = On у php.ini або ignore_user_abort(true) у початковому коді worker-скрипта (коді перед його циклом запитів, із якого починаються деякі приклади воркерів) — це базове значення для кожного запиту, тож кожен потік, що обслуговується там, поводиться так, ніби сам зробив цей виклик, і потребує перевірки connection_aborted().

Використання нативного flush()

Нативна PHP-функція flush() теж працює для потокової передачі, але спершу потрібно очистити всі шари буферизації виводу. Краще використовуйте oxphp_stream_flush() — вона сама керує буферами виводу й інтегрована із системою зворотного тиску OxPHP.

php

<?php
header('Content-Type: text/event-stream');
header('Cache-Control: no-cache');

while (ob_get_level()) {
    ob_end_clean();
}

for ($i = 0; $i < 100; $i++) {
    echo "data: " . json_encode(['counter' => $i]) . "\n\n";
    flush();
    sleep(1);
}

SSE у режимі воркерів

SSE працює і в стандартному режимі, і в режимі воркерів. У режимі воркерів потік займає свій робочий потік лише доти, доки його скрипт виконується або чекає, не призупиняючи свій файбер, як це робить flush, поки буфер на 64 фрагменти заповнений. Поки потік чекає в oxphp_sleep() або на читанні через хук середовища виконання streams, робочий потік обслуговує інші запити; один робочий потік виконує до 256 запитів одночасно (див. Мультиплексування файберів).

У трансляції (broadcast) кожен відкритий потік має отримати кожну подію, тож приклад читає їх із Redis Stream, а не знімає зі списку. Читання Redis Stream лишає його записи на місці, а кожен читач зберігає власну позицію, тож кожен клієнт читає кожну подію, яку Redis Stream ще містить. BRPOP на списку працює навпаки: він видаляє елемент, який повертає, і цей елемент дістається рівно одному зі з’єднань, що чекають на списку. Така форма підходить для черги завдань і не підходить для трансляції: за двох відкритих потоків кожна подія дійшла б лише до одного з них.

php

<?php
require __DIR__ . '/../vendor/autoload.php';

oxphp_worker(function () {
    if (($_SERVER['HTTP_ACCEPT'] ?? '') !== 'text/event-stream') {
        http_response_code(400);
        echo json_encode(['error' => 'SSE only']);
        return;
    }

    header('Content-Type: text/event-stream');
    header('Cache-Control: no-cache');

    // A stream outlives max_execution_time, which is 30 s unless the PHP configuration sets another value
    set_time_limit(0);

    // One connection per stream: the blocking read below holds it for as long as the
    // stream lives. The last argument is the read timeout, which has to exceed the
    // 25 s the read blocks for.
    $redis = new Redis();
    $redis->connect('redis', 6379, 2.0, null, 0, 30.0);

    // The newest entry in the stream. It also finds out whether the key holds a stream at all,
    // while the answer can still be an error status: once the first frame is out the response is
    // a 200, and a browser reconnects after every close of one.
    $tail = $redis->xRevRange('events', '+', '-', 1);
    if ($tail === false) {
        error_log('SSE: ' . $redis->getLastError());
        http_response_code(500);
        return;
    }

    // Resume after the last event the browser saw (it sends the id back in Last-Event-ID).
    // A first connection, or a header that is not a stream id, starts after the newest entry.
    $last = $_SERVER['HTTP_LAST_EVENT_ID'] ?? '';
    if (!preg_match('/^\d{1,19}-\d{1,19}$/', $last)) {
        $last = $tail ? array_key_first($tail) : '0-0';
    }

    // Send the headers now; a quiet stream would otherwise send none until its first read returns.
    // This named event carries the starting id, which the browser takes as its Last-Event-ID before
    // the first event arrives; onmessage does not fire for it.
    echo "event: ready\nid: {$last}\ndata:\n\n";
    oxphp_stream_flush();

    while (true) {
        // Reading leaves the entries in the stream, so every open stream gets every one
        $read = $redis->xRead(['events' => $last], 100, 25000);
        if ($read === false) {
            // Redis refused the read, the key having been replaced by a non-stream, say. End the stream;
            // the browser reconnects and the check above answers it with an error status
            return;
        }
        foreach ($read['events'] ?? [] as $id => $fields) {
            echo "id: {$id}\ndata: {$fields['data']}\n\n";
            $last = $id;
        }
        if (!$read) {
            // No entry within 25 s — send heartbeat to keep the connection alive
            echo ": heartbeat\n\n";
        }
        oxphp_stream_flush();
    }
});

Публікуйте з будь-якого PHP-процесу (обробника запиту, воркера черги, CLI-скрипта):

php

$redis->xAdd('events', '*', ['data' => json_encode($payload)], 1000, true);

Останні два аргументи обмежують Redis Stream приблизно 1000 записами (MAXLEN ~ 1000). Як data надсилайте JSON або будь-що інше без переносу рядка: приклад записує це значення в один рядок data:.

На що спирається приклад:

Ідентифікатор запису — це ідентифікатор події. Ідентифікатор на кшталт 1790976033892-0 надсилається як id: події, під час перепідключення браузер повертає останній побачений у Last-Event-ID, і потік продовжується після нього (див. Поведінка під час зупинки).
Перший фрейм — це подія, що несе початковий ідентифікатор. Safari і Firefox беруть id: як Last-Event-ID браузера лише з фрейму, у якому є рядок data:, тож за голого id: браузер, з’єднання якого обірвалося до першої події, не мав би ідентифікатора, щоб його надіслати: він перепідключився б, почав би з найновішого запису й пропустив би все, що було опубліковано в проміжку. Подія ready з порожнім рядком data: дає йому такий ідентифікатор. onmessage не спрацьовує для іменованої події, а addEventListener('ready', …) спрацьовує; клієнтська бібліотека, яка передає кожну подію в один колбек, має цю подію пропускати.
Потік, який не може стартувати, отримує статус помилки ще до першого фрейму. Redis відмовляється читати ключ, що містить щось інше, ніж Redis Stream (наприклад, список, що лишився від попередньої черги), і жоден ідентифікатор не змусить таке читання спрацювати, тож приклад з’ясовує це першою ж командою, записує в лог повідомлення Redis і відповідає 500. Це має бути саме статус, а не потік, який просто завершується: щойно перший фрейм надіслано, відповідь має статус 200, а браузер перепідключається після кожного закриття потоку з 200 — кожні кілька секунд, доки сторінка відкрита, щоразу з новим з’єднанням із Redis і без жодного запису в лозі. Після статусу помилки EventSource не повторює спроб; сторінка отримує подію error з readyState, що дорівнює EventSource.CLOSED, і, якщо їй потрібне нове з’єднання, має сама відкрити новий EventSource. Те саме стосується недосяжного Redis: connect() кидає виняток, відповідь — 500, і EventSource закрито остаточно. Читання, від якого Redis відмовляється вже після старту потоку (скажімо, ключ замінили на щось інше, ніж Redis Stream), завершує потік через return, і перепідключення проходить через перевірку вище. Без цього return цикл сприйняв би false як тайм-аут і надсилав би самі лише heartbeat, один за одним.
Last-Event-ID перевіряється на формат. Заголовок надходить від клієнта й може містити що завгодно, а Redis не приймає значення, яке не є ідентифікатором Redis Stream, — наприклад, abc. Приклад обробляє таке значення як відсутній заголовок і починає після найновішого запису, як і під час першого підключення. Шаблон допускає до 19 цифр з кожного боку дефіса, тож кожне прийняте ним значення лишається в діапазоні, який читає Redis.
Новий потік зчитує найновіший ідентифікатор один раз. Він не передає $ у кожному виклику: Redis розв’язує $ у найновіший ідентифікатор на момент кожного виклику, тож записи, додані між двома викликами, пропускаються.
Історія обмежена. Читач, чию останню подію вже обрізано (браузер, який був від’єднаний, або потік, затриманий повільним клієнтом, поки опублікували більше записів, ніж дозволяє ліміт), продовжує з найстарішого запису, що ще лишився. Записи, обрізані в проміжку, не доставляються, і ніщо про це не повідомляє. Підбирайте ліміт так, щоб він покривав найдовший перебій, після якого потік має відновитися.
Кожен потік має власне з’єднання з тайм-аутом читання, більшим за час блокування. Без аргументу тайм-ауту phpredis використовує default_socket_timeout, і читання, що блокується довше, завершується помилкою RedisException. Не відкривайте одне з’єднання на воркер, щоб віддавати його кожному потоку: з увімкненими хуками середовища виконання streams воркер виконує інші запити, поки потік чекає на читанні, а запит, який скористався з’єднанням, утримує його до свого завершення. Другий потік, що дістається того самого з’єднання, чекає на перший. Якщо перший усе ще відкритий, коли очікування спливає (межу задають max_execution_time і default_socket_timeout; див. розділ про хуки), phpredis кидає RedisException, і другий потік завершується помилкою.
Скрипт вимикає таймер виконання PHP. max_execution_time дорівнює 30 секундам, якщо конфігурація PHP не задає іншого значення, а релізний образ не містить php.ini; потік же має пережити цей ліміт. Якщо таймер лишити ввімкненим, скрипт завершиться фатальною помилкою: таймер спрацьовує через 30 секунд і набирає чинності, коли повертається поточне читання, тож із 25-секундними читаннями, як тут, тихий потік завершується на 50-й секунді. У режимі воркерів виклик вимикає ліміт лише запиту самого потоку й жодного іншого; див. Режим воркерів → Дедлайни запитів.
Клієнта, який закрив з’єднання, помічають на наступному flush. До того його запит, а з ним і з’єднання з Redis, лишається відкритим: тут щонайбільше 25 с — час блокування, який водночас є інтервалом heartbeat.

Redis Pub/Sub теж доставляє кожне повідомлення кожному підключеному підписникові, але нічого не зберігає: повідомлення, опубліковане, поки клієнт перепідключається, втрачено, і немає ідентифікатора, з якого можна відновитися. Використовуйте його лише там, де пропущена подія не має значення.

Кілька екземплярів за балансувальником навантаження

Потік — це одне довге з’єднання з одним екземпляром, який балансувальник обирає один раз. Якщо екземплярів більше одного:

Події мають переходити між екземплярами. Об’єкти OxPHP\Shared\* живуть усередині одного процесу (див. Спільний стан), тож потік на одному екземплярі не може через них побачити подію, опубліковану на іншому. Обидва мають читати те саме зовнішнє сховище — саме ним і є Redis Stream вище.
Місткість рахується для кожного екземпляра окремо. Кожен екземпляр обслуговує стільки потоків, скільки вміщує його власний пул воркерів (див. Рекомендації). Віддавайте перевагу методу балансування, який рахує відкриті з’єднання, як-от least_conn у nginx, а не round-robin: round-robin розподіляє з’єднання за моментом надходження, а не за тим, як довго вони лишаються відкритими, а потік лишається відкритим набагато довше, ніж звичайний запит.
Липкі сесії (sticky sessions) не потрібні. Браузер, що перепідключається, може потрапити на інший екземпляр, і це нормально, якщо перший фрейм і кожна подія несуть id:, а всі екземпляри читають той самий Redis Stream.
Проксі не має буферизувати потік. Проксі з буферизацією затримує події, доки не заповниться його буфер. У nginx вимкніть буферизацію для location потоку через proxy_buffering off або надсилайте зі скрипта X-Accel-Buffering: no, який nginx враховує, якщо proxy_ignore_headers не вказує інакше.
Heartbeat має надходити частіше, ніж спливає тайм-аут простою балансувальника. Див. «Проміжні проксі закривають неактивні SSE-з’єднання» в розділі Усунення несправностей.
Використовуйте HTTP/2 між браузером і балансувальником. Через HTTP/1.1 браузер дозволяє близько шести відкритих з’єднань на хост для всіх своїх вкладок разом, і кожен потік займає одне з них; HTTP/2 передає багато потоків через одне з’єднання.

Усунення несправностей

Клієнт не отримує даних, доки скрипт не завершиться

Буфер виводу PHP перехоплює вивід замість того, щоб передавати його потоком. Так буває, коли активні шари OB, а oxphp_stream_flush() не викликається.

Виправлення: викликайте oxphp_stream_flush() після кожної події. Ця функція скидає всі шари буферизації виводу PHP і надсилає накопичений вивід як фрагмент.

SSE-з’єднання закриваються через кілька хвилин

Спрацьовує PHP-обмеження max_execution_time і зупиняє скрипт. SSE-потоки мають працювати довше за налаштований ліміт.

Виправлення: вимкніть таймер виконання для окремого запиту на початку потокового скрипта:

php

set_time_limit(0);

Це краще, ніж глобально встановлювати max_execution_time = 0, — ліміт лишається чинним для ендпоінтів без SSE. Або ж, якщо весь екземпляр призначений лише для довготривалих потоків:

php.ini

; php.ini
max_execution_time = 0

Проміжні проксі закривають неактивні SSE-з’єднання

Балансувальники навантаження й проксі часто закривають з’єднання, якими 30–60 секунд не передаються дані.

Виправлення: регулярно надсилайте heartbeat-коментар, щоб з’єднання лишалося активним:

php

echo ": heartbeat\n\n";
oxphp_stream_flush();

oxphp_stream_flush() повертає false

Раніше в цьому самому запиті було викликано oxphp_finish_request(). Коли відповідь завершено, потокова передача вже неможлива. Перевірте, чи немає в коді випадкових викликів oxphp_finish_request() до початку потокової передачі.

Приклад для Docker

Для SSE-ендпоінтів таймер виконання PHP має бути вимкнений або встановлений на велике значення. Поза режимом воркерів, як у цьому прикладі, кожне активне SSE-з’єднання займає один PHP-воркер на весь час існування потоку, тож розмір пулу воркерів має враховувати очікувану кількість одночасних потоків.

compose.yaml

services:
  app:
    image: ghcr.io/oxphp/oxphp:0.13.0
    ports:
      - "8080:80"
    volumes:
      - ./src:/var/www/html
    environment:
      DOCUMENT_ROOT: "/var/www/html/public"
      ENTRY_FILE: "index.php"
      PHP_WORKERS: "32"

Кожен потоковий скрипт має викликати set_time_limit(0) на початку, щоб таймер запиту не спрацював посеред потоку. Так глобальний max_execution_time лишається чинним для запитів без SSE, і в режимі воркерів теж: там кожен запит має власний ліміт, тож потік, вимикаючи свій, лишає обмеженими запити, що обслуговуються поруч із ним (див. Режим воркерів → Дедлайни запитів).

Примітки

Потоки подій ніколи не стискаються, і на це є два незалежні правила: text/event-stream не належить до стисних медіатипів, а відповідь, довжина якої невідома на момент надсилання заголовків, передається без стиснення незалежно від того, що приймає клієнт. Це стосується кожного кодування, яке пропонує OxPHP (Brotli, Zstandard і gzip однаково), тож жодне налаштування не ввімкне стиснення для потоку.
oxphp_stream_flush() повертає false, якщо в цьому самому запиті вже було викликано oxphp_finish_request().
У режимі воркерів потік займає свій робочий потік лише доти, доки виконується або чекає, не призупиняючи свій файбер; поки він чекає в oxphp_sleep() або на перехопленому хуком читанні, робочий потік обробляє інші запити. В інших режимах воркер лишається зайнятим, доки PHP-скрипт не завершиться.

Поведінка під час зупинки

Коли сервер отримує SIGTERM (поетапне розгортання, docker stop, виселення пода в Kubernetes), він коректно завершує роботу, дочікуючись поточних запитів:

Кожен відкритий SSE-потік чисто завершується на своєму наступному flush — його обробник виходить так, ніби з’єднання закрилося, тож колбеки register_shutdown_function() усе одно виконуються, а error_get_last()['message'] містить Request cancelled (shutdown).
Клієнти HTTP/2 отримують фрейм GOAWAY; keep-alive-з’єднання HTTP/1.1 закриваються. Браузерний EventSource перепідключається автоматично — до справного екземпляра, якщо перед групою серверів стоїть балансувальник навантаження.
Потік не отримує 503: його заголовки 200 уже надіслано, тож статус переписати неможливо. Проєктуйте клієнти так, щоб вони відновлювалися з останнього ідентифікатора події (надсилайте id: з кожною подією; під час перепідключення браузер повертає його в заголовку запиту Last-Event-ID, доступному в PHP як $_SERVER['HTTP_LAST_EVENT_ID']). Саме так і робить приклад для режиму воркерів.

Звичайні (не потокові) запити, що виконуються в момент SIGTERM, не перериваються: вони отримують усе вікно дренажу, щоб нормально завершитися, і скасовуються лише ті запити, які ще виконуються, коли спливає DRAIN_TIMEOUT_SECONDS (за замовчуванням 25), — з приблизно ще 2 секундами на згортання. Установіть період очікування завершення (termination grace period) в оркестраторі більшим за DRAIN_TIMEOUT_SECONDS + 2 плюс найдовший блокувальний виклик, у якому запит може перебувати в момент дедлайну (див. Коректне завершення роботи).

«Потокова» тут означає будь-яку відповідь, яка вже скинула chunked-вивід, — сервер не може відрізнити скінченне потокове завантаження від нескінченного потоку подій, тож велике завантаження, що вже скидало дані й виконується в момент SIGTERM, теж завершується достроково, не лише SSE. Запит, який викликав oxphp_finish_request() до SIGTERM, вважається звичайним: його відповідь завершена, а решта фонової роботи отримує вікно дренажу.

Обробник, заблокований усередині одного довготривалого нативного виклику (нативний sleep(), важкий preg_match, блокувальний запит до бази даних), не можна перервати, доки цей виклик не повернеться. У режимі воркерів у потокових циклах краще використовуйте кооперативний oxphp_sleep() — він передає керування планувальнику файберів і прокидається одразу під час зупинки. Поза режимом воркерів oxphp_sleep() зводиться до звичайного блокувального очікування, тож потік реагує на зупинку на своєму наступному flush після повернення зі sleep.

Промпти для вашого агента

Ці промпти розраховані на агента з підключеним MCP-сервером документації OxPHP — потрібні сторінки він прочитає сам.

prompt

Побудуй ендпоінт Server-Sent Events у цьому проєкті на OxPHP. Прочитай сторінку `features/sse` через MCP-сервер документації OxPHP.
Напиши обробник навколо даних, які цей репозиторій уже має передавати, — heartbeat, відключення клієнтів, відновлення з останнього ідентифікатора події — і скажи, скільки воркерів знадобиться для очікуваного мною рівня одночасності.

Дивіться також

Режим воркерів — постійні PHP-воркери для зменшення накладних витрат на ініціалізацію
Тайм-аути — налаштування або вимкнення тайм-ауту запиту для довготривалих з’єднань
Функції PHP — повний довідник з oxphp_stream_flush() і oxphp_is_streaming()
Стиснення — які відповіді стискаються і яким кодуванням