Server-Sent Events (SSE)

OxPHP diffuse des données en temps réel vers les clients via le protocole Server-Sent Events, avec une contre-pression intégrée. Définissez Content-Type: text/event-stream dans votre script PHP et appelez oxphp_stream_flush(). OxPHP se charge du reste.

Fonctionnement

Une réponse en streaming suit un cycle de vie prévisible :

Définir les en-têtes. Votre script PHP définit Content-Type: text/event-stream via header() et écrit des lignes au format SSE avec echo.
Le premier flush ouvre le flux. Le premier appel à oxphp_stream_flush() envoie les en-têtes HTTP au client et bascule en mode streaming. La connexion cliente reste ouverte.
Chaque flush envoie un chunk. Chaque appel suivant à oxphp_stream_flush() vide la sortie mise en tampon sous forme d'un nouveau chunk, le livrant immédiatement au client.
La contre-pression s'active lorsque le tampon se remplit. OxPHP maintient un tampon interne pouvant contenir jusqu'à 64 chunks entre le worker PHP et le client. Lorsque le tampon est plein — parce qu'un client lent n'a pas consommé les chunks précédents — oxphp_stream_flush() se bloque jusqu'à ce que de la place se libère. Cela évite une croissance illimitée de la mémoire.
Nettoyage à la sortie ou à la déconnexion. Lorsque le script PHP se termine, OxPHP ferme la connexion de manière gracieuse. Si le client se déconnecte en cours de flux, OxPHP détecte le canal fermé au prochain flush, met le drapeau connection_aborted() de PHP à true et arme un abandon gracieux — les boucles portables qui vérifient connection_aborted() sortent proprement par leur chemin de terminaison normal, tandis que les boucles qui ne le vérifient pas sont malgré tout arrêtées par un abandon implicite au flush suivant.

Note

Gardez des charges utiles d'événements réduites pour maintenir un débit régulier. Les charges utiles volumineuses peuvent remplir rapidement le tampon de 64 chunks, ce qui provoque un blocage de PHP à chaque flush.

Exemples

Flux SSE de base

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

Vérifier l'état du streaming

Utilisez oxphp_is_streaming() pour vérifier si la requête courante est déjà en mode streaming. C'est utile dans un middleware ou des gestionnaires de requêtes partagés :

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

Détecter les déconnexions client

Les boucles SSE de longue durée doivent vérifier connection_aborted() pour s'interrompre proprement lorsque le client ferme la connexion. Cela correspond à l'idiome standard PHP / php-fpm et permet au script d'exécuter toute logique de nettoyage (fermeture des handles de base de données, libération des verrous, exécution des blocs finally) avant de sortir :

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

Si le script ne vérifie jamais connection_aborted(), OxPHP le termine tout de même via un abandon implicite au flush suivant la déconnexion du client, mais les blocs finally situés après des chemins de code qui contournent l'appel au flush risquent de ne pas s'exécuter. Privilégiez la vérification explicite pour le code qui détient des ressources externes.

Utiliser flush() natif

Le flush() natif de PHP fonctionne aussi pour le streaming, mais il faut d'abord vider toutes les couches de tampon de sortie. Préférez oxphp_stream_flush() — il gère automatiquement les tampons de sortie et s'intègre au système de contre-pression d'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 en mode worker

SSE fonctionne aussi bien en mode standard qu'en mode worker. En mode worker, la connexion en streaming occupe le worker pendant toute la durée du flux. Le worker ne traite la requête suivante qu'une fois le script terminé.

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

Dépannage

Le client ne reçoit aucune donnée avant la fin du script

Le tampon de sortie PHP capture la sortie au lieu de la diffuser. Cela se produit lorsque des couches OB sont actives et que oxphp_stream_flush() n'est pas appelé.

Correctif : appelez oxphp_stream_flush() après chaque événement. Cette fonction vide toutes les couches de tampon de sortie PHP et envoie la sortie accumulée sous forme de chunk.

Les connexions SSE sont fermées après quelques minutes

Le max_execution_time de PHP se déclenche et termine le script. Les flux SSE doivent s'exécuter plus longtemps que la limite configurée.

Correctif : désactivez le minuteur d'exécution par requête en tête du script de streaming :

php

set_time_limit(0);

C'est préférable à la définition globale de max_execution_time = 0 — cela laisse la limite en place pour les endpoints non-SSE. Autrement, si l'instance entière est dédiée aux flux de longue durée :

php.ini

; php.ini
max_execution_time = 0

Des proxys intermédiaires ferment les connexions SSE inactives

Les répartiteurs de charge et les proxys ferment souvent les connexions qui ne transportent aucune donnée pendant 30 à 60 secondes.

Correctif : envoyez un heartbeat sous forme de commentaire à intervalles réguliers pour maintenir la connexion active :

php

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

oxphp_stream_flush() renvoie false

oxphp_finish_request() a été appelé plus tôt dans la même requête. Une fois la réponse terminée, le streaming n'est plus possible. Vérifiez votre code à la recherche d'appels involontaires à oxphp_finish_request() avant le début du streaming.

Exemple Docker

Les endpoints SSE nécessitent que le minuteur d'exécution de PHP soit désactivé ou réglé sur une valeur élevée. Chaque connexion SSE active occupe un worker PHP pendant toute la durée du flux ; dimensionnez donc le pool de workers pour absorber le nombre de flux concurrents attendu.

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"

Chaque script de streaming doit appeler set_time_limit(0) en tête afin que le minuteur par requête ne se déclenche pas en cours de flux. Cela maintient le max_execution_time global en vigueur pour les requêtes non-SSE.

Bonnes pratiques

Utilisez oxphp_stream_flush() plutôt que le flush() natif pour une gestion automatique des tampons de sortie et une intégration à la contre-pression.
Envoyez des heartbeats périodiques sous forme de commentaires (: heartbeat\n\n) toutes les 20 à 30 secondes pour empêcher les proxys intermédiaires de fermer les connexions inactives et pour détecter tôt les déconnexions client.
Gardez des charges utiles d'événements réduites. Les charges utiles volumineuses remplissent plus vite le tampon de 64 chunks, ce qui fait caler PHP à chaque flush. Pour de gros volumes de données, envoyez un ID d'événement et laissez le client récupérer la charge utile complète via une requête distincte.
Désactivez le minuteur d'exécution de PHP par script avec set_time_limit(0) pour les endpoints SSE de longue durée, ou réglez max_execution_time sur une valeur suffisamment élevée pour couvrir la durée maximale de flux attendue.
Dimensionnez votre pool de workers pour le pic de flux concurrents. Chaque connexion SSE active occupe un worker PHP pendant toute sa durée. Prévoyez au moins un worker par client concurrent attendu, plus des workers supplémentaires pour les requêtes non-SSE ordinaires.

Remarques

La compression Brotli est automatiquement ignorée pour les réponses en streaming. La compression ne s'applique qu'aux réponses entièrement mises en tampon.
oxphp_stream_flush() renvoie false si oxphp_finish_request() a déjà été appelé sur la même requête.
En mode worker, le worker reste occupé pendant toute la durée du flux et ne traite la requête suivante qu'une fois le script PHP terminé.

Comportement à l'arrêt

Lorsque le serveur reçoit SIGTERM (un déploiement progressif, un docker stop, une éviction de pod Kubernetes), il draine gracieusement :

Chaque flux SSE ouvert est terminé proprement à son prochain flush — son gestionnaire abandonne comme si la connexion s'était fermée, de sorte que les callbacks de register_shutdown_function() s'exécutent toujours et que error_get_last()['message'] contient Request cancelled (shutdown).
Les clients HTTP/2 reçoivent une trame GOAWAY ; les connexions HTTP/1.1 keep-alive sont fermées. L'EventSource du navigateur se reconnecte automatiquement — vers une instance saine lorsqu'un répartiteur de charge est placé devant la flotte.
Le flux ne reçoit pas de 503 : ses en-têtes 200 ont déjà été envoyés, le statut ne peut donc pas être réécrit. Concevez les clients pour qu'ils reprennent à partir du dernier id d'événement (envoyez id: avec chaque événement ; à la reconnexion, le navigateur le renvoie dans l'en-tête de requête Last-Event-ID, disponible en PHP via $_SERVER['HTTP_LAST_EVENT_ID']).

Les requêtes ordinaires (non-streaming) en cours au moment du SIGTERM ne sont pas interrompues : elles disposent de toute la fenêtre de drainage pour se terminer normalement, et seules les requêtes encore en cours lorsque DRAIN_TIMEOUT_SECONDS (25 par défaut) expire sont annulées, avec environ 2 secondes de plus pour se dérouler. Réglez la période de grâce de terminaison de l'orchestrateur au-dessus de DRAIN_TIMEOUT_SECONDS + 2.

Le terme « streaming » désigne ici toute réponse ayant déjà vidé une sortie en chunks — le serveur ne peut pas distinguer un téléchargement en streaming fini d'un flux d'événements infini, si bien qu'un gros téléchargement vidé en cours au moment du SIGTERM est lui aussi terminé prématurément, pas seulement le SSE. Une requête ayant appelé oxphp_finish_request() avant le SIGTERM est comptée comme ordinaire : sa réponse est complète, et son travail de fond restant bénéficie de la fenêtre de drainage.

Un gestionnaire bloqué à l'intérieur d'un unique appel natif de longue durée (un sleep() natif, un preg_match lourd, une requête de base de données bloquante) ne peut pas être interrompu tant que cet appel n'est pas revenu. En mode worker, préférez le oxphp_sleep() coopératif dans les boucles de streaming — il rend la main à l'ordonnanceur de fibers et est réveillé immédiatement à l'arrêt. Hors mode worker, oxphp_sleep() retombe sur un sleep bloquant classique, de sorte que le flux réagit à l'arrêt à son prochain flush après le retour du sleep.

Voir aussi

Mode worker — processus PHP persistants pour réduire le coût de démarrage
Délais d'expiration — configurer ou désactiver le délai d'expiration des requêtes pour les connexions de longue durée
Fonctions PHP — référence complète de oxphp_stream_flush() et oxphp_is_streaming()
Compression — comportement de la compression Brotli et quelles réponses sont compressées