Server-Sent Events (SSE)

OxPHP przesyła strumieniowo dane w czasie rzeczywistym do klientów za pośrednictwem protokołu Server-Sent Events, z wbudowanym przeciwciśnieniem. Ustaw Content-Type: text/event-stream w swoim skrypcie PHP i wywołaj oxphp_stream_flush(). Resztą zajmie się OxPHP.

Jak to działa

Odpowiedź strumieniowa przechodzi przez przewidywalny cykl życia:

Ustaw nagłówki. Twój skrypt PHP ustawia Content-Type: text/event-stream za pomocą header() i zapisuje wiersze w formacie SSE, używając echo.
Pierwsze opróżnienie bufora otwiera strumień. Pierwsze wywołanie oxphp_stream_flush() wysyła nagłówki HTTP do klienta i przechodzi w tryb strumieniowania. Połączenie z klientem pozostaje otwarte.
Każde opróżnienie bufora wysyła porcję. Każde kolejne wywołanie oxphp_stream_flush() opróżnia zbuforowane dane wyjściowe jako nową porcję, dostarczając ją natychmiast do klienta.
Przeciwciśnienie włącza się, gdy bufor się zapełni. OxPHP utrzymuje wewnętrzny bufor mieszczący do 64 porcji między workerem PHP a klientem. Gdy bufor jest pełny — ponieważ wolny klient nie odebrał wcześniejszych porcji — oxphp_stream_flush() blokuje się do momentu zwolnienia miejsca. Zapobiega to nieograniczonemu wzrostowi zużycia pamięci.
Sprzątanie przy zakończeniu lub rozłączeniu. Gdy skrypt PHP kończy działanie, OxPHP łagodnie zamyka połączenie. Jeśli klient rozłączy się w trakcie strumienia, skrypt dowiaduje się o tym przy pierwszym opróżnieniu bufora po tym, jak OxPHP zauważy zamknięcie połączenia. OxPHP ustawia flagę connection_aborted() w PHP i, o ile skrypt nie wywołał ignore_user_abort(true), zatrzymuje skrypt wewnątrz tego opróżnienia: wywołanie nie wraca, i właśnie tak PHP zatrzymuje skrypt pod PHP-FPM, gdy jego danych wyjściowych nie da się już zapisać. Skrypt, który wywołał ignore_user_abort(true), odzyskuje sterowanie po opróżnieniu i działa dalej, a jego pętlę kończy sprawdzenie connection_aborted() (zobacz Wykrywanie rozłączeń klienta).

Note

Utrzymuj małe ładunki zdarzeń, aby zachować płynną przepustowość. Duże ładunki mogą szybko zapełnić 64-porcjowy bufor, powodując blokowanie PHP przy każdym opróżnieniu bufora.

Przykłady

Podstawowy strumień 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();
    }
}

Sprawdzanie stanu strumieniowania

Użyj oxphp_is_streaming(), aby sprawdzić, czy bieżące żądanie jest już w trybie strumieniowania. Przydaje się to w middleware lub we współdzielonych obsługach żądań:

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();

Wykrywanie rozłączeń klienta

Strumień, który wysłał już nagłówki, dowiaduje się o odejściu klienta dopiero wtedy, gdy opróżnia bufor. Domyślnie to właśnie w tym opróżnieniu skrypt się zatrzymuje: oxphp_stream_flush() (lub flush()) nie wraca, więc warunek while (!connection_aborted()) nie zostaje już nigdy ponownie sprawdzony, a każdy blok finally, wewnątrz którego znajdowało się opróżnienie, zostaje pominięty. PHP-FPM w ten sam sposób zatrzymuje skrypt, którego danych wyjściowych nie da się już zapisać.

Wykonuje się jeszcze zakończenie żądania: funkcje zwrotne register_shutdown_function(), w których connection_aborted() zwraca 1, oraz destruktory obiektów znikających wraz z żądaniem. W odróżnieniu od PHP-FPM, który odrzuca wszystko, co one zapiszą, OxPHP zatrzymuje funkcję zwrotną przy pierwszych wysłanych przez nią danych wyjściowych i pomija funkcje zwrotne zarejestrowane po niej. Destruktory nadal się wykonują, a pierwszy z nich, który wyśle dane wyjściowe, zostaje zatrzymany w ten sam sposób, a destruktory przypadające po nim są pomijane. Dane wyjściowe przechowywane w buforze wyjściowym (otwartym przez ob_start() albo w tym, który otwiera output_buffering w php.ini) są wysyłane dopiero wtedy, gdy ten bufor zostanie opróżniony. Zadbaj, by takie sprzątanie nie generowało danych wyjściowych, albo wywołaj ignore_user_abort(true), jak poniżej.

Aby posprzątać w samym skrypcie, w bloku finally lub po pętli, wywołaj ignore_user_abort(true) i pozwól, by pętlę zakończyło connection_aborted(). Opróżnienie, które wykryje odejście klienta, wtedy wraca, connection_aborted() od tej chwili zwraca 1, a pętla kończy się poprzez swój warunek:

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)` podtrzymuje działanie strumienia

Przy tym wywołaniu jedyną rzeczą, która kończy pętlę, jest sprawdzenie connection_aborted(): skrypt działa dalej po rozłączeniu klienta, to, co zapisuje, jest odrzucane, a connection_aborted() zgłasza rozłączenie. To ten sam kontrakt co w każdym innym SAPI. Taka pętla potrzebuje własnego ograniczenia: strumień, który poprosił o to, by przeżyć swojego klienta, a przy tym ani nie sprawdza connection_aborted(), ani nie kończy się sam, zajmuje swojego workera aż do zamknięcia serwera.

Wywołanie dotyczy tylko żądania, które je wykonuje. Tak samo liczy się jednak wartość, z którą żądanie startuje: ignore_user_abort = On w php.ini albo ignore_user_abort(true) w kodzie inicjalizującym skryptu workera (kodzie przed jego pętlą żądań, od którego zaczynają się niektóre przykłady workerów) to punkt wyjścia każdego żądania, więc każdy obsługiwany tam strumień zachowuje się tak, jakby sam wykonał to wywołanie, i potrzebuje swojego sprawdzenia connection_aborted().

Korzystanie z natywnej funkcji flush()

Natywna funkcja flush() z PHP również nadaje się do strumieniowania, ale najpierw wymaga wyczyszczenia wszystkich warstw bufora wyjściowego. Lepiej używać oxphp_stream_flush() — automatycznie zarządza buforami wyjściowymi i integruje się z systemem przeciwciśnienia 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 w trybie worker

SSE działa zarówno w trybie standardowym, jak i w trybie worker. W trybie worker strumień zajmuje swój wątek workera tylko wtedy, gdy jego skrypt działa albo czeka bez zawieszania swojego Fibera, tak jak opróżnienie bufora czeka, gdy 64-porcjowy bufor jest pełny. Gdy strumień czeka w oxphp_sleep() albo na odczyt przechodzący przez hook runtime streams, wątek obsługuje inne żądania; wątek workera obsługuje jednocześnie do 256 żądań (zobacz Multipleksowanie Fiberów).

W rozgłaszaniu każdy otwarty strumień musi otrzymać każde zdarzenie, dlatego przykład odczytuje je ze strumienia Redis (Redis Stream), zamiast zdejmować je z listy. Odczyt strumienia Redis pozostawia jego wpisy na miejscu, a każdy czytelnik utrzymuje własną pozycję, więc każdy klient odczytuje każde zdarzenie, które strumień jeszcze przechowuje. BRPOP na liście działa odwrotnie: usuwa zwracany element, a ten element trafia do dokładnie jednego z połączeń czekających na tej liście. To właściwy kształt dla kolejki zadań i niewłaściwy dla rozgłaszania: przy dwóch otwartych strumieniach każde zdarzenie dotarłoby tylko do jednego z nich.

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();
    }
});

Publikuj z dowolnego procesu PHP (handlera żądania, workera kolejki, skryptu CLI):

php

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

Dwa ostatnie argumenty ograniczają strumień do około 1000 wpisów (MAXLEN ~ 1000). Jako data wysyłaj JSON albo cokolwiek innego bez znaku nowej linii: przykład zapisuje tę wartość w jednym wierszu data:.

Na czym opiera się przykład:

Identyfikator wpisu jest identyfikatorem zdarzenia. Identyfikator taki jak 1790976033892-0 wychodzi jako id: zdarzenia, przeglądarka przy ponownym połączeniu odsyła ostatni, jaki widziała, w Last-Event-ID, a strumień jest kontynuowany od miejsca po nim (zobacz Zachowanie przy zamknięciu).
Pierwsza ramka to zdarzenie niosące identyfikator początkowy. Safari i Firefox przyjmują id: jako Last-Event-ID przeglądarki tylko z ramki, która ma wiersz data:, więc samo id: zostawiłoby przeglądarkę odciętą przed pierwszym zdarzeniem bez identyfikatora do wysłania: połączyłaby się ponownie, zaczęła od najnowszego wpisu i przegapiła wszystko, co opublikowano w międzyczasie. Zdarzenie ready z pustym wierszem data: daje jej ten identyfikator. onmessage nie uruchamia się dla nazwanego zdarzenia, a addEventListener('ready', …) tak; biblioteka kliencka, która przekazuje każde zdarzenie do jednej funkcji zwrotnej, musi je pominąć.
Strumień, który nie może wystartować, dostaje status błędu przed swoją pierwszą ramką. Redis odmawia odczytu klucza, który przechowuje coś innego niż strumień (na przykład listę pozostałą po wcześniejszej kolejce), i żaden identyfikator nie sprawi, że taki odczyt się powiedzie, więc przykład wykrywa to swoim pierwszym poleceniem, zapisuje w logu komunikat zwrócony przez Redis i odpowiada 500. Musi to być status, a nie strumień, który po prostu się kończy: gdy pierwsza ramka zostanie wysłana, odpowiedź ma status 200, a przeglądarka łączy się ponownie po każdym zamknięciu strumienia 200, co kilka sekund, dopóki strona pozostaje otwarta, za każdym razem z nowym połączeniem do Redis i bez żadnego śladu w logu. EventSource nie ponawia prób po statusie błędu; strona dostaje zdarzenie error z readyState równym EventSource.CLOSED i musi sama otworzyć nowy EventSource, jeśli go potrzebuje. To samo dotyczy Redis, z którym nie można się połączyć: connect() rzuca wyjątek, odpowiedź ma status 500, a EventSource zostaje zamknięty na dobre. Odczyt, którego Redis odmówi już po starcie strumienia (na przykład gdy klucz zostanie zastąpiony czymś, co nie jest strumieniem), kończy strumień instrukcją return, a ponowne połączenie przechodzi przez powyższe sprawdzenie. Bez tego return pętla wzięłaby false za upływ limitu czasu i wysyłałaby wyłącznie heartbeaty, jeden za drugim.
Format Last-Event-ID jest sprawdzany. Nagłówek pochodzi od klienta i może zawierać cokolwiek, a Redis odrzuca wartość, która nie jest identyfikatorem strumienia, na przykład abc. Przykład traktuje taką wartość jak brak nagłówka i zaczyna po najnowszym wpisie, tak jak przy pierwszym połączeniu. Wzorzec dopuszcza do 19 cyfr po każdej stronie myślnika, dzięki czemu każda akceptowana wartość mieści się w zakresie, który Redis odczytuje.
Nowy strumień odczytuje najnowszy identyfikator raz. Nie przekazuje $ przy każdym wywołaniu: Redis rozwiązuje $ na najnowszy identyfikator w chwili każdego wywołania, więc wpisy dodane między dwoma wywołaniami zostają pominięte.
Historia jest ograniczona. Czytelnik, którego ostatnie zdarzenie zostało w międzyczasie usunięte przy przycinaniu (przeglądarka, która była rozłączona, albo strumień wstrzymany przez wolnego klienta, podczas gdy opublikowano więcej wpisów, niż wynosi limit), kontynuuje od najstarszego wpisu, który jeszcze pozostał. Wpisy przycięte w międzyczasie nie są dostarczane i nic tego nie zgłasza. Dobierz limit tak, aby pokrywał najdłuższą przerwę, po której strumień ma móc się wznowić.
Każdy strumień ma własne połączenie z limitem czasu odczytu dłuższym niż czas blokowania. Bez argumentu limitu czasu phpredis używa default_socket_timeout, a odczyt, który blokuje dłużej, kończy się wyjątkiem RedisException. Nie otwieraj jednego połączenia na workera, by przekazywać je każdemu strumieniowi: przy włączonych hookach runtime streams worker obsługuje inne żądania, gdy strumień czeka na odczyt, a żądanie, które użyło połączenia, zatrzymuje je aż do swojego końca. Drugi strumień, który sięgnie po to samo połączenie, czeka na pierwszy. Jeśli pierwszy jest nadal otwarty, gdy czas oczekiwania minie (granica wynika z max_execution_time i default_socket_timeout; zobacz sekcję o hookach), phpredis rzuca RedisException i drugi strumień kończy się niepowodzeniem.
Skrypt wyłącza licznik czasu wykonania PHP. max_execution_time wynosi 30 sekund, chyba że konfiguracja PHP ustawia inną wartość, a obraz wydania nie zawiera php.ini; strumień ma działać dłużej. Przy pozostawionym liczniku skrypt kończy się błędem krytycznym: licznik zadziała po 30 sekundach i odniesie skutek, gdy trwający odczyt zwróci sterowanie, więc przy 25-sekundowych odczytach z tego przykładu cichy strumień kończy się w 50. sekundzie. W trybie worker wywołanie wyłącza limit wyłącznie żądania samego strumienia i żadnego innego; zobacz Tryb worker → Limity czasu żądań.
Klient, który zamknął połączenie, zostaje zauważony przy następnym opróżnieniu bufora. Do tego czasu jego żądanie, a wraz z nim połączenie do Redis, pozostaje otwarte: tutaj najwyżej 25 s, czyli czas blokowania, który jest jednocześnie odstępem między heartbeatami.

Redis Pub/Sub również dostarcza każdą wiadomość każdemu podłączonemu subskrybentowi, ale niczego nie przechowuje: wiadomość opublikowana, gdy klient łączy się ponownie, przepada, a nie ma identyfikatora, od którego można by wznowić. Używaj go tylko tam, gdzie pominięte zdarzenie nie ma znaczenia.

Kilka instancji za load balancerem

Strumień to jedno długie połączenie z jedną instancją, wybraną raz przez load balancer. Gdy instancji jest więcej niż jedna:

Zdarzenia muszą przechodzić między instancjami. Obiekty OxPHP\Shared\* żyją wewnątrz jednego procesu (zobacz Stan współdzielony), więc strumień na jednej instancji nie zobaczy przez nie zdarzenia opublikowanego na innej. Obie muszą czytać z tego samego zewnętrznego magazynu, a tym właśnie jest powyższy strumień Redis.
Pojemność liczy się per instancja. Każda instancja obsługuje tyle strumieni, na ile pozwala jej własna pula workerów (zobacz Dobre praktyki). Zamiast round-robin wybierz metodę równoważenia, która liczy otwarte połączenia, taką jak least_conn w nginx: round-robin rozkłada połączenia według kolejności nadejścia, a nie według tego, jak długo pozostają otwarte, a strumień jest otwarty znacznie dłużej niż zwykłe żądanie.
Sticky sessions nie są potrzebne. Przeglądarka łącząca się ponownie może trafić na inną instancję, co nie stanowi problemu, o ile pierwsza ramka i każde zdarzenie niosą id:, a każda instancja czyta ten sam strumień.
Proxy nie może buforować strumienia. Buforujące proxy wstrzymuje zdarzenia, dopóki jego bufor się nie zapełni. W nginx wyłącz buforowanie dla location strumienia przez proxy_buffering off albo wysyłaj ze skryptu X-Accel-Buffering: no, który nginx respektuje, chyba że proxy_ignore_headers stanowi inaczej.
Heartbeaty muszą być częstsze niż limit bezczynności load balancera. Zobacz „Pośrednie proxy zamykają bezczynne połączenia SSE” w sekcji Rozwiązywanie problemów.
Między przeglądarką a load balancerem używaj HTTP/2. Przez HTTP/1.1 przeglądarka pozwala na około sześć otwartych połączeń na host we wszystkich swoich kartach, a każdy strumień zajmuje jedno z nich; HTTP/2 przenosi wiele strumieni przez jedno połączenie.

Rozwiązywanie problemów

Klient nie otrzymuje żadnych danych, dopóki skrypt się nie zakończy

Bufor wyjściowy PHP przechwytuje dane wyjściowe zamiast je strumieniować. Dzieje się tak, gdy warstwy OB są aktywne, a oxphp_stream_flush() nie jest wywoływane.

Rozwiązanie: Wywołuj oxphp_stream_flush() po każdym zdarzeniu. Ta funkcja opróżnia wszystkie warstwy bufora wyjściowego PHP i wysyła nagromadzone dane wyjściowe jako porcję.

Połączenia SSE są zamykane po kilku minutach

Zadziałał max_execution_time z PHP i zakończył skrypt. Strumienie SSE muszą działać dłużej niż skonfigurowany limit.

Rozwiązanie: Wyłącz licznik czasu wykonania dla danego żądania na początku skryptu strumieniującego:

php

set_time_limit(0);

Jest to preferowane względem globalnego ustawienia max_execution_time = 0 — pozostawia limit w mocy dla endpointów innych niż SSE. Alternatywnie, jeśli cała instancja jest przeznaczona do długo żyjących strumieni:

php.ini

; php.ini
max_execution_time = 0

Pośrednie proxy zamykają bezczynne połączenia SSE

Load balancery i proxy często zamykają połączenia, przez które nie przepływają żadne dane przez 30–60 sekund.

Rozwiązanie: Wysyłaj heartbeat w formie komentarza w regularnych odstępach czasu, aby utrzymać aktywne połączenie:

php

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

oxphp_stream_flush() zwraca false

oxphp_finish_request() zostało wywołane wcześniej w tym samym żądaniu. Po zakończeniu odpowiedzi strumieniowanie nie jest możliwe. Sprawdź w swoim kodzie, czy nie ma przypadkowych wywołań oxphp_finish_request() przed rozpoczęciem strumieniowania.

Przykład z Dockerem

Endpointy SSE wymagają wyłączenia licznika czasu wykonania PHP lub ustawienia go na wysoką wartość. Poza trybem worker, tak jak w tym przykładzie, każde aktywne połączenie SSE zajmuje jednego workera PHP przez cały czas trwania strumienia, więc dobierz rozmiar puli workerów tak, aby pomieściła oczekiwaną liczbę równoczesnych strumieni.

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"

Każdy skrypt strumieniujący powinien wywołać set_time_limit(0) na początku, aby licznik czasu dla danego żądania nie zadziałał w trakcie strumienia. Dzięki temu globalny max_execution_time pozostaje w mocy dla żądań innych niż SSE, również w trybie worker: tam każde żądanie ma własny limit, więc strumień wyłączający swój limit nie zdejmuje ograniczenia z żądań obsługiwanych obok niego (zobacz Tryb worker → Terminy żądań).

Dobre praktyki

Używaj oxphp_stream_flush() zamiast natywnej funkcji flush(), aby uzyskać automatyczne zarządzanie buforem wyjściowym i integrację z przeciwciśnieniem.
Wysyłaj okresowe heartbeaty w formie komentarza (: heartbeat\n\n) co 20–30 sekund, aby pośrednie proxy nie zamykały bezczynnych połączeń i aby wcześnie wykrywać rozłączenia klientów.
Utrzymuj małe ładunki zdarzeń. Duże ładunki szybciej zapełniają 64-porcjowy bufor, powodując zacinanie się PHP przy każdym opróżnieniu bufora. W przypadku dużych danych wyślij identyfikator zdarzenia i pozwól klientowi pobrać pełny ładunek za pomocą osobnego żądania.
Wyłącz licznik czasu wykonania PHP dla poszczególnych skryptów za pomocą set_time_limit(0) w przypadku długo żyjących endpointów SSE lub ustaw max_execution_time na tyle wysoko, aby pokryć najdłuższy oczekiwany czas trwania strumienia.
Dobierz rozmiar puli workerów do szczytowej liczby równoczesnych strumieni. Poza trybem worker, a w trybie worker w przypadku strumienia, którego oczekiwania nie zawieszają jego Fibera, każde aktywne połączenie SSE zajmuje jednego workera PHP przez cały czas swojego trwania (zobacz SSE w trybie worker). W tych przypadkach zaplanuj co najmniej jednego workera na każdego oczekiwanego równoczesnego klienta oraz dodatkowych workerów na zwykłe żądania inne niż SSE.

Uwagi

Strumienie zdarzeń nigdy nie są kompresowane, i to na mocy dwóch niezależnych reguł: text/event-stream nie jest kompresowalnym typem mediów, a odpowiedź, której długość nie jest znana w chwili wysłania nagłówków, przechodzi bez kompresji niezależnie od tego, co akceptuje klient. Dotyczy to każdego kodowania oferowanego przez OxPHP (zarówno Brotli, jak i Zstandard oraz gzip), więc żadna konfiguracja nie włączy kompresji dla strumienia.
oxphp_stream_flush() zwraca false, jeśli oxphp_finish_request() zostało już wywołane w tym samym żądaniu.
W trybie worker strumień zajmuje swój wątek workera tylko wtedy, gdy działa lub czeka bez zawieszania swojego Fibera; gdy czeka w oxphp_sleep() albo na odczyt objęty hookiem, wątek obsługuje inne żądania. W pozostałych trybach worker pozostaje zajęty aż do zakończenia skryptu PHP.

Zachowanie przy zamknięciu

Gdy serwer otrzyma SIGTERM (kroczące wdrożenie, docker stop, eksmisja poda w Kubernetes), łagodnie kończy pracę, wygaszając połączenia:

Każdy otwarty strumień SSE jest czysto kończony przy swoim następnym flush — jego handler przerywa działanie tak, jakby połączenie zostało zamknięte, więc funkcje zarejestrowane przez register_shutdown_function() nadal się uruchamiają, a error_get_last()['message'] zawiera Request cancelled (shutdown).
Klienci HTTP/2 otrzymują ramkę GOAWAY; połączenia keep-alive HTTP/1.1 są zamykane. EventSource w przeglądarce łączy się ponownie automatycznie — z działającą instancją, gdy przed flotą stoi load balancer.
Strumień nie otrzymuje odpowiedzi 503: jego nagłówki 200 zostały już wysłane, więc statusu nie da się nadpisać. Projektuj klientów tak, aby wznawiali od ostatniego identyfikatora zdarzenia (wysyłaj id: z każdym zdarzeniem; przy ponownym połączeniu przeglądarka odsyła go z powrotem w nagłówku żądania Last-Event-ID, dostępnym w PHP jako $_SERVER['HTTP_LAST_EVENT_ID']). Dokładnie to robi przykład z trybem worker.

Zwykłe (niestrumieniowe) żądania będące w toku w momencie SIGTERM nie są przerywane: mają całe okno wygaszania na normalne zakończenie, a anulowane zostają tylko te żądania, które nadal działają po upływie DRAIN_TIMEOUT_SECONDS (domyślnie 25), przy czym dostają jeszcze ~2 sekundy na dokończenie. Ustaw okres karencji przy zakończeniu (termination grace period) orkiestratora powyżej DRAIN_TIMEOUT_SECONDS + 2, powiększonego o najdłuższe blokujące wywołanie, w którym żądanie może się znajdować w chwili upływu terminu (zobacz Łagodne zamknięcie).

„Strumieniowanie" oznacza tu każdą odpowiedź, która już opróżniła fragmentowane dane wyjściowe — serwer nie potrafi odróżnić skończonego pobierania strumieniowego od nieskończonego strumienia zdarzeń, więc duże opróżnione pobieranie będące w toku w momencie SIGTERM również zostaje zakończone wcześniej, nie tylko SSE. Żądanie, które wywołało oxphp_finish_request() przed SIGTERM, liczy się jako zwykłe: jego odpowiedź jest kompletna, a pozostała praca w tle otrzymuje okno wygaszania.

Handler zablokowany wewnątrz jednego długo trwającego wywołania natywnego (natywny sleep(), kosztowny preg_match, blokujące zapytanie do bazy danych) nie może zostać przerwany, dopóki to wywołanie nie zwróci sterowania. W trybie worker w pętlach strumieniujących preferuj kooperacyjny oxphp_sleep() — oddaje sterowanie planiście Fiberów i zostaje wybudzony natychmiast przy zamknięciu. Poza trybem worker oxphp_sleep() przechodzi na zwykły blokujący sen, więc strumień zareaguje na zamknięcie przy swoim następnym flush po powrocie z uśpienia.

Prompty dla Twojego agenta

Te prompty zakładają agenta z podłączonym serwerem MCP dokumentacji OxPHP — sam przeczyta strony, których potrzebuje.

prompt

Zbuduj w tym projekcie endpoint Server-Sent Events na OxPHP. Przeczytaj stronę `features/sse` przez serwer MCP dokumentacji OxPHP.
Napisz handler wokół danych, które to repozytorium musi już wypychać — heartbeaty, rozłączenia klienta, wznawianie od ostatniego identyfikatora zdarzenia — i powiedz mi, ilu workerów będzie wymagać współbieżność, której się spodziewam.

Zobacz także

Tryb worker — trwałe workery PHP zmniejszające narzut na inicjalizację
Limity czasu — konfigurowanie lub wyłączanie limitu czasu żądania dla długo żyjących połączeń
Funkcje PHP — pełna dokumentacja oxphp_stream_flush() i oxphp_is_streaming()
Kompresja — które odpowiedzi są kompresowane i jakim kodowaniem