Server-Sent Events (SSE)

OxPHP передаёт клиентам данные в реальном времени по протоколу Server-Sent Events со встроенным противодавлением (backpressure). Задайте Content-Type: text/event-stream в своём PHP-скрипте и вызовите oxphp_stream_flush(). Остальное OxPHP берёт на себя.

Как это работает

Потоковый ответ проходит через предсказуемый жизненный цикл:

Задайте заголовки. Ваш PHP-скрипт устанавливает Content-Type: text/event-stream через header() и пишет строки в формате SSE с помощью echo.
Первый сброс открывает поток. Первый вызов oxphp_stream_flush() отправляет клиенту HTTP-заголовки и переводит запрос в потоковый режим. Соединение с клиентом остаётся открытым.
Каждый сброс отправляет чанк. Каждый последующий вызов oxphp_stream_flush() сбрасывает буферизованный вывод в виде нового чанка, немедленно доставляя его клиенту.
Противодавление включается при заполнении буфера. OxPHP поддерживает внутренний буфер объёмом до 64 чанков между PHP-воркером и клиентом. Когда буфер заполнен — из-за того, что медленный клиент не забрал предыдущие чанки — oxphp_stream_flush() блокируется, пока не освободится место. Это предотвращает неограниченный рост потребления памяти.
Очистка при завершении или отключении. Когда PHP-скрипт завершается, OxPHP корректно закрывает соединение. Если клиент отключается посреди потока, скрипт узнаёт об этом при первом сбросе после того, как OxPHP увидит закрытие соединения. OxPHP устанавливает флаг PHP connection_aborted() и, если скрипт не вызывал ignore_user_abort(true), останавливает скрипт внутри этого сброса: вызов не возвращает управление — именно так PHP под PHP-FPM останавливает скрипт, вывод которого больше невозможно записать. Скрипт, вызвавший ignore_user_abort(true), получает управление обратно из сброса и продолжает работать, а цикл завершает проверка connection_aborted() (см. Обнаружение отключений клиента).

Note

Держите полезную нагрузку событий небольшой, чтобы сохранять ровную пропускную способность. Крупная полезная нагрузка может быстро заполнить 64-чанковый буфер, из-за чего PHP будет блокироваться на каждом сбросе.

Примеры

Базовый 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();

Обнаружение отключений клиента

Поток, который уже отправил заголовки, узнаёт об уходе клиента только при сбросе. По умолчанию на этом сбросе скрипт и останавливается: oxphp_stream_flush() (или flush()) не возвращает управление, поэтому условие while (!connection_aborted()) больше не вычисляется, а все блоки finally, внутри которых находился сброс, пропускаются. 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(). Тогда сброс, обнаруживший уход клиента, возвращает управление, 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) в начальной загрузке скрипта воркера (коде перед его циклом запросов, с которого начинаются некоторые примеры воркеров) — это исходное состояние каждого запроса, поэтому каждый поток, обслуживаемый там, ведёт себя так, будто сделал этот вызов, и ему нужна проверка connection_aborted().

Использование нативного flush()

Нативный flush() из PHP тоже работает для потоковой передачи, но сначала требует очистить все слои буфера вывода. Предпочитайте 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 работает как в стандартном режиме, так и в режиме воркеров. В режиме воркеров SSE-поток удерживает свой поток воркера только пока его скрипт выполняется или ждёт, не приостанавливая свой файбер, — так ждёт сброс, пока 64-чанковый буфер заполнен. Пока SSE-поток ждёт в oxphp_sleep() или на чтении через хук времени выполнения streams, поток воркера обслуживает другие запросы; один поток воркера выполняет до 256 запросов одновременно (см. Мультиплексирование файберов).

При широковещательной рассылке каждый открытый поток должен получить каждое событие, поэтому пример читает их из 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:.

На что опирается пример:

Id записи — это id события. Id вида 1790976033892-0 уходит как id: события, браузер при переподключении возвращает последний увиденный id в Last-Event-ID, и поток продолжается после него (см. Поведение при завершении работы).
Первый кадр — событие, несущее начальный id. Safari и Firefox принимают id: в качестве Last-Event-ID браузера только из кадра со строкой data:, поэтому голый id: оставил бы браузер, отключившийся до первого события, без id для отправки: он переподключился бы, начал с самой новой записи и пропустил всё, что было опубликовано в промежутке. Событие ready с пустой строкой data: даёт ему такой id. onmessage не срабатывает для именованного события, а addEventListener('ready', …) срабатывает; клиентская библиотека, передающая все события в один колбэк, должна это событие пропускать.
Поток, который не может начаться, получает статус ошибки до своего первого кадра. Redis отказывается читать ключ, в котором хранится не Redis Stream (например, список, оставшийся от прежней очереди), и никакой id не заставит такое чтение сработать, поэтому пример выясняет это первой же командой, записывает в лог сообщение 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 отвергает значение, которое не является id записи Redis Stream, например abc. Пример обращается с таким значением как с отсутствующим заголовком и начинает после самой новой записи, как при первом подключении. Шаблон допускает до 19 цифр по обе стороны от дефиса, поэтому каждое принятое значение остаётся в диапазоне, который читает Redis.
Новый поток читает самый новый id один раз. Он не передаёт $ при каждом вызове: Redis разрешает $ в самый новый id на момент каждого вызова, поэтому записи, добавленные между двумя вызовами, пропускаются.
История ограничена. Читатель, чьё последнее событие уже обрезано (браузер, который был отключён, или поток, задержанный медленным клиентом, пока публиковалось больше записей, чем допускает лимит), продолжает с самой старой из оставшихся записей. Записи, обрезанные в промежутке, не доставляются, и ничто об этом не сообщает. Подбирайте лимит так, чтобы он покрывал самый долгий перерыв, после которого поток должен возобновиться.
У каждого потока своё соединение с таймаутом чтения больше времени блокировки. Без аргумента таймаута 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-й секунде. В режиме воркеров вызов отключает лимит только собственного запроса потока и ничей больше; см. Режим воркеров → Дедлайны запросов.
Клиент, закрывший соединение, обнаруживается при следующем сбросе. До тех пор его запрос, а вместе с ним и соединение с Redis, остаётся открытым: здесь не дольше 25 с — времени блокировки, которое одновременно служит интервалом heartbeat.

Redis Pub/Sub тоже доставляет каждое сообщение каждому подключённому подписчику, но ничего не хранит: сообщение, опубликованное, пока клиент переподключается, пропадает, и id для возобновления нет. Используйте его только там, где пропущенное событие не имеет значения.

Несколько экземпляров за балансировщиком нагрузки

Поток — это одно долгое соединение с одним экземпляром, который балансировщик выбирает один раз. Когда экземпляров больше одного:

События должны переходить между экземплярами. Объекты OxPHP\Shared\* живут внутри одного процесса (см. Разделяемое состояние), поэтому поток на одном экземпляре не может через них увидеть событие, опубликованное на другом. Оба должны читать одно и то же внешнее хранилище — именно этим и служит Redis Stream выше.
Ёмкость считается для каждого экземпляра отдельно. Каждый экземпляр обслуживает столько потоков, на сколько оставляет места его собственный пул воркеров (см. Рекомендации). Предпочитайте метод балансировки, учитывающий открытые соединения, например least_conn в nginx, а не round-robin: round-robin распределяет соединения по моменту поступления, а не по тому, как долго они остаются открытыми, а поток остаётся открытым гораздо дольше обычного запроса.
Sticky-сессии не нужны. Переподключившийся браузер может попасть на другой экземпляр, и это нормально, пока первый кадр и каждое событие несут 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-соединения закрываются через несколько минут

Срабатывает max_execution_time в PHP и завершает скрипт. 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() уже был вызван в том же запросе.
В режиме воркеров SSE-поток занимает свой поток воркера только пока выполняется или ждёт, не приостанавливая свой файбер; пока он ждёт в 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 события (отправляйте 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 после возврата из сна.

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

Эти промпты рассчитаны на агента с подключённым MCP-сервером документации OxPHP — нужные страницы он прочитает сам.

prompt

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

Смотрите также

Режим воркеров — постоянные PHP-воркеры для снижения накладных расходов на инициализацию
Таймауты — настройка или отключение таймаута запроса для долгоживущих соединений
Функции PHP — полный справочник по oxphp_stream_flush() и oxphp_is_streaming()
Сжатие — какие ответы сжимаются и каким кодированием