Server-Sent Events (SSE)

OxPHP は、組み込みのバックプレッシャーを備えた Server-Sent Events プロトコルで、クライアントにリアルタイムデータをストリーミングします。PHP スクリプトで Content-Type: text/event-stream を設定し、oxphp_stream_flush() を呼び出すだけです。残りは OxPHP が処理します。

仕組み

ストリーミングレスポンスは、予測可能なライフサイクルに沿って進みます。

ヘッダーを設定する。 PHP スクリプトは header() で Content-Type: text/event-stream を設定し、echo を使って SSE 形式の行を書き出します。
最初のフラッシュでストリームが開く。 oxphp_stream_flush() の最初の呼び出しで HTTP ヘッダーがクライアントに送信され、ストリーミングモードに入ります。クライアント接続は開いたままになります。
各フラッシュがチャンクを送る。 以降の oxphp_stream_flush() の呼び出しごとに、バッファされた出力が新しいチャンクとしてフラッシュされ、即座にクライアントへ届けられます。
バッファが埋まるとバックプレッシャーが働く。 OxPHP は PHP ワーカーとクライアントの間に最大 64 チャンクの内部バッファを保持します。バッファが満杯になると（遅いクライアントが先行するチャンクをまだ消費していないため）、oxphp_stream_flush() は空きができるまでブロックします。これによりメモリの際限ない増加を防ぎます。
終了時または切断時のクリーンアップ。 PHP スクリプトが終了すると、OxPHP は接続をグレースフルにクローズします。クライアントがストリームの途中で切断した場合、OxPHP は次のフラッシュでチャンネルが閉じられたことを検知し、PHP の 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() を使用します。これはミドルウェアや共有のリクエストハンドラーで役立ちます。

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() を使う

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 は標準モードとワーカーモードの両方で動作します。ワーカーモードでは、ストリーミング接続がストリームの全期間にわたってワーカーを占有します。ワーカーは、スクリプトが終了して初めて次のリクエストを処理します。

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 接続が閉じられる

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 秒間データを運ばない接続をしばしば閉じます。

対処: 一定間隔でコメントハートビートを送信し、接続をアクティブに保ってください。

php

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

oxphp_stream_flush() が false を返す

同じリクエスト内で先に oxphp_finish_request() が呼び出されています。レスポンスがいったん終了すると、ストリーミングはできません。ストリーミング開始前に oxphp_finish_request() を意図せず呼び出していないか、コードを確認してください。

Docker の例

SSE エンドポイントでは、PHP の実行タイマーを無効化するか、高い値に設定する必要があります。アクティブな SSE 接続はそれぞれ、ストリームの全期間にわたって PHP ワーカーを 1 つ占有するため、想定される同時ストリーム数に合わせてワーカープールのサイズを決めてください。

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) を呼び出すべきです。これにより、非 SSE リクエストに対してはグローバルな max_execution_time が有効なまま維持されます。

ベストプラクティス

ネイティブの flush() ではなく oxphp_stream_flush() を使用する。 出力バッファの自動管理とバックプレッシャー統合が得られます。
定期的にコメントハートビートを送信する（: heartbeat\n\n）。20〜30 秒ごとに送ることで、中間プロキシがアイドル状態の接続を閉じるのを防ぎ、クライアントの切断を早期に検知できます。
イベントのペイロードを小さく保つ。 ペイロードが大きいと 64 チャンクのバッファが早く埋まり、フラッシュのたびに PHP が停滞します。大きなデータの場合は、イベント ID を送り、完全なペイロードは別のリクエストでクライアントに取得させてください。
スクリプトごとに PHP の実行タイマーを無効化する。 長時間稼働する SSE エンドポイントでは set_time_limit(0) を使うか、想定される最長のストリーム時間をカバーできるよう max_execution_time を十分高く設定してください。
ピーク時の同時ストリーム数に合わせてワーカープールのサイズを決める。 アクティブな SSE 接続はそれぞれ、その全期間にわたって PHP ワーカーを 1 つ保持します。想定される同時クライアントごとに少なくとも 1 つのワーカーを見込み、加えて通常の非 SSE リクエスト用のワーカーを追加してください。

注意点

ストリーミングレスポンスでは Brotli 圧縮が自動的にスキップされます。圧縮は完全にバッファされたレスポンスにのみ適用されます。
同じリクエストで oxphp_finish_request() がすでに呼び出されていた場合、oxphp_stream_flush() は false を返します。
ワーカーモードでは、ワーカーはストリームの全期間にわたって占有され続け、PHP スクリプトが終了して初めて次のリクエストを処理します。

シャットダウン時の挙動

サーバーが SIGTERM を受信すると（ローリングデプロイ、docker stop、Kubernetes の Pod 退避など）、グレースフルにドレインします。

開いている各 SSE ストリームは、次の flush でクリーンに終了されます。ハンドラーは接続が閉じられたかのように中断するため、register_shutdown_function() のコールバックは依然として実行され、error_get_last()['message'] は Request cancelled (shutdown) を返します。
HTTP/2 クライアントは GOAWAY フレームを受信します。HTTP/1.1 の keep-alive 接続は閉じられます。ブラウザの EventSource は自動的に再接続します。フリートの前段にロードバランサーがある場合は、健全なインスタンスへ再接続します。
ストリームは 503 を受け取りません。その 200 ヘッダーはすでに送信済みなので、ステータスを書き換えることはできません。クライアントは最後のイベント ID から再開するよう設計してください（各イベントに id: を送信します。再接続時、ブラウザはそれを Last-Event-ID リクエストヘッダーとして送り返し、PHP では $_SERVER['HTTP_LAST_EVENT_ID'] として利用できます）。

SIGTERM 時に処理中だった通常の（ストリーミングではない）リクエストは中断されません。ドレインウィンドウの全時間を使って通常どおり完了でき、DRAIN_TIMEOUT_SECONDS（デフォルト 25）が経過してもまだ実行中のリクエストだけがキャンセルされ、後始末のためにさらに約 2 秒が与えられます。オーケストレーターの終了猶予期間は DRAIN_TIMEOUT_SECONDS + 2 より大きく設定してください。

ここでの「ストリーミング」とは、すでにチャンク化された出力をフラッシュしたあらゆるレスポンスを指します。サーバーは有限のストリーミングダウンロードと無限のイベントストリームを区別できないため、SIGTERM 時に処理中だった大きなフラッシュ済みダウンロードも、SSE だけでなく早期に終了されます。SIGTERM より前に oxphp_finish_request() を呼び出したリクエストは通常のものとして扱われます。そのレスポンスは完了しており、残りのバックグラウンド処理はドレインウィンドウを与えられます。

単一の長時間稼働するネイティブ呼び出し（ネイティブの sleep()、重い preg_match、ブロッキングなデータベースクエリ）の内部でブロックしているハンドラーは、その呼び出しが返るまで中断できません。ワーカーモードでは、ストリーミングループで協調的な oxphp_sleep() を優先してください。これはファイバースケジューラーに制御を譲り、シャットダウン時には即座に起こされます。ワーカーモード以外では oxphp_sleep() は通常のブロッキングスリープにフォールバックするため、ストリームはスリープが返った後の次の flush でシャットダウンに反応します。