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 обнаруживает закрытый канал при следующем сбросе, устанавливает флаг connection_aborted() в 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();

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

Долгоживущие SSE-циклы должны проверять connection_aborted(), чтобы чисто выйти, когда клиент закрывает соединение. Это соответствует стандартной идиоме PHP / php-fpm и позволяет скрипту выполнить любую логику очистки (закрытие дескрипторов базы данных, освобождение блокировок, завершение блоков finally) перед выходом:

php

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

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

try {
    while (!connection_aborted()) {
        echo "data: " . json_encode(['ts' => time()]) . "\n\n";
        oxphp_stream_flush();
        sleep(1);
    }
} finally {
    $db = null; // runs on normal exit AND on connection_aborted exit
}

Note

Если скрипт никогда не проверяет connection_aborted(), OxPHP всё равно завершит его через неявный аварийный выход при следующем сбросе после отключения клиента, но блоки finally, следующие за путями выполнения, которые обходят вызов сброса, могут не выполниться. Для кода, удерживающего внешние ресурсы, предпочитайте явную проверку.

Использование нативного 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 работает как в стандартном режиме, так и в режиме воркеров. В режиме воркеров потоковое соединение занимает воркер на всё время работы потока. Воркер обрабатывает следующий запрос только после того, как скрипт завершится.

php

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

$redis = new Redis();
$redis->pconnect('redis', 6379);

oxphp_worker(function () use ($redis) {
    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');

    while (true) {
        $message = $redis->brPop('events', 25);
        if ($message) {
            echo "data: {$message[1]}\n\n";
        } else {
            // No message within timeout — send heartbeat to keep the connection alive
            echo ": heartbeat\n\n";
        }
        oxphp_stream_flush();
    }
});

Устранение неполадок

Клиент не получает данные, пока скрипт не завершится

Буфер вывода 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.10.0
    ports:
      - "8080:8080"
    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.

Примечания

Brotli-сжатие автоматически пропускается для потоковых ответов. Сжатие применяется только к полностью буферизованным ответам.
oxphp_stream_flush() возвращает false, если oxphp_finish_request() уже был вызван в том же запросе.
В режиме воркеров воркер остаётся занятым на всё время работы потока и обрабатывает следующий запрос только после выхода из 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 после возврата из сна.

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

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