Utilisation concrète des SSE (Server-Sent Events) en PHP

Dans mon précédent article, j’ai expliqué ce que sont les SSE et comment on peut les utiliser de manière minimale.

Dans l’exemple que je donnais, le serveur envoyait au client des données qu’il générait de lui-même. Mais dans la vraie vie, le serveur va vouloir envoyer au client des données qui ont été générées par d’autres clients. L’exemple habituel, c’est celui des salons de discussion en temps réel : dès qu’un utilisateur écrit un message, on veut qu’il apparaisse sur les écrans des autres utilisateurs.

Les SSE ne sont donc qu’une composante de la solution technique à mettre en place. Voyons maintenant quelle est l’architecture logicielle globale à mettre en place pour créer un système de chat simple. Bien évidemment, plusieurs solutions sont peuvent être utilisées, nous allons voir l’une d’elles, avec une implémentation en PHP basée sur le framework Temma.

Architecture logicielle

L’interface web va être très simple : une liste de messages (qui affichera pour chaque message le nom de son expéditeur et son contenu textuel), et un formulaire pour envoyer des messages (avec un champ pour le nom d’expéditeur et un champ pour le texte du message).
Les messages envoyés seront retournés aux différents clients par SSE.

La question est de savoir comment les différents contrôleurs qui envoient des SSE sont eux-mêmes avertis qu’il y a de nouveaux messages à transmettre.
Il va falloir mettre en place un système pour transmettre les données. On pourrait regarder du côté des files de messages et des traitements asynchrones (voir un précédent article sur le sujet), mais ce serait une erreur : ce dont on a besoin ici, ce n’est pas de traiter des tâches de manière asynchrone, c’est de transmettre rapidement un message à plusieurs clients en même temps.

Il serait possible de stocker les messages en base de données. Chaque contrôleur pourrait ensuite interroger régulièrement la base pour regarder s’il y a de nouveaux messages. Mais plus il y aurait de clients connectés et plus il y aurait de requêtes sur la base, et ces accès incessants tueraient les performances de la base. Sans compter que ce ne serait plus du temps réel, il y aurait un lag à cause du délai entre deux “polling” sur la base.

Nous allons plutôt partir sur une solution à base de communication réseau entre les contrôleurs et un “broker”, un élément central qui reçoit les messages et les redistribue à tous les contrôleurs. Pour le faire facilement, nous utiliserons la bibliothèque réseau ZeroMQ.
J’ai déjà parlé plusieurs fois de ZeroMQ sur ce blog. C’est une bibliothèque qui permet de créer très facilement des échanges entre processus ou entre threads, avec une abstraction supérieure de celle des sockets réseau habituelles.

Cela nous donnerait quelque chose comme ça :

À l’écriture d’un message, le client envoie les données en AJAX au contrôleur Message.
Le contrôleur ouvre une socket ZeroMQ vers le broker de type PUSH, pour lui envoyer le message.
Le broker était en lecture bloquante sur sa socket entrante PULL. Il reçoit le message, et le renvoie sur sa socket sortante de type PUB (publication).
Tous les contrôleurs Event, qui se connectent automatiquement à cette dernière socket ZeroMQ en mode SUB (subscription), reçoivent le message et le renvoient par SSE aux clients qui leur sont connectés.

Fichier de configuration

Nous avons donc quatre sockets ZeroMQ différentes. Elles vont toutes être listées dans le fichier de configuration, comme des sources de données. Les connexions ne sont réellement effectuées qu’au moment où elles sont utilisées pour la première fois, donc cela ne pose pas de problème.

Voici le contenu du fichier de configuration etc/temma.json :

{
  "application": {
    "dataSources": {
      "zmq_client_push": "zmq://PUSH@127.0.0.1:5000",
      "zmq_broker_pull": "zmq-bind://PULL@127.0.0.1:5000",
      "zmq_broker_pub": "zmq-bind://PUB@127.0.0.1:6000",
      "zmq_client_sub": "zmq://SUB@127.0.0.1:6000"
    },
    "enableSessions": false
  }
}Langage du code : JSON / JSON avec commentaires (json)

Les quatre sockets ZeroMQ sont donc listées comme des sources de données, et seront donc utilisables dans les différents contrôleurs gérés par Temma :

zmq_client_push : utilisée par le contrôleur Message pour envoyer au broker le message qu’il a reçu.
zmq_broker_pull : utilisée par le broker pour recevoir les messages envoyés par les contrôleurs Message.
zmq_broker_pub : utilisée par le broker pour redistribuer les messages reçus vers tous les contrôleurs Event.
zmq_client_sub : utilisée par le contrôleur Event pour recevoir les messages publiés par le broker.

En plus de cela, les sessions sont désactivées, car nous n’en aurons pas besoin.

Code du broker

Le broker est un script en ligne de commande géré par Comma (la partie de Temma qui s’occupe des scripts CLI). Comme expliqué dans la documentation de Comma, les scripts CLI s’écrivent globalement comme des contrôleurs web, avec juste quelques spécificités.

Voici le code du fichier cli/Broker.php :

<?php

/** Contrôleur de script broker. */
class Broker extends \Temma\Web\Controller {
    /** Action principale. */
    public function run() {
        // activation de la socket
        $this->zmq_broker_pub->connect();
        // boucle infinie
        while (true) {
            // lecture d'un nouveau message envoyé par le contrôleur Message
            $msg = $this->zmq_broker_pull[''];
            // envoi du message aux contrôleurs Event
            $this->zmq_broker_pub[''] = $msg;
        }
    }
}
Langage du code : PHP (php)

Le code est très simple :

Ligne 4 : Définition du contrôleur Broker.
Ligne 6 : Définition de l’action run.
Ligne 8 : Ouverture explicite de la socket ZeroMQ PUB (publication). En temps normal, cela n’est pas nécessaire, les sockets ZeroMQ sont ouvertes au moment où elles sont utilisées pour la première fois. Là c’est nécessaire pour que les contrôleurs Event qui tenteront de s’y connecter trouvent une socket qui accepte les connexions.
Ligne 10 : Boucle infinie.
Ligne 12 : Lecture bloquante sur la socket ZeroMQ PULL. Avec Temma, on lit et on écrit dans les sources de données comme si c’était des tableaux associatifs. Ici, la clé est une chaîne vide.
Ligne 14 : Écriture sur la socket ZeroMQ PUB du message qui vient d’être reçu. Là encore, on manipule la source de données comme un tableau associatif, avec une chaîne vide comme clé.

Le script se lance simplement avec la commande :

bin/comma Broker runLangage du code : Bash (bash)

Code du contrôleur Message

Ce contrôleur est appelé en AJAX. Il reçoit deux paramètres POST :

from : le nom de l’expéditeur du message.
text : le texte du message.

Il envoie ces données sur sa socket ZeroMQ PUSH, et il retourne toujours la valeur true sérialisée en JSON.

Voici le contenu du fichier controller/Message.php :

<?php

use \Temma\Attributes\View as TµView;

/** Contrôleur Message. */
class Message extends \Temma\Web\Controller {
    /** Action qui reçoit les nouveaux messages. */
    #[TµView('~Json')]
    public function broadcast() {
        // définition de la valeur de retour
        $this['json'] = true;
        // récupération des paramètres POST
        $from = $_POST['from'] ?? null;
        $text = $_POST['text'] ?? null;
        // vérification des données
        if (!trim($from) || !trim($text))
            return;
        // envoi des données vers le broker
        $this->zmq_client_push[''] = [
            'from' => $from,
            'text' => $text,
        ];
    }
}
Langage du code : PHP (php)

Ligne 6 : Définition du contrôleur Message, qui est un contrôleur web standard.
Ligne 8 : Utilisation de l’attribut \Temma\Attributes\View pour que ce soit la vue JSON qui soit exécutée après le traitement de l’action (au lieu de la vue Smarty par défaut).
Ligne 9 : Définition de l’action broadcast.
Ligne 11 : Définition de la valeur de retour (true) qui sera sérialisée en JSON.
Lignes 13 et 14 : Récupération des paramètres POST.
Lignes 16 et 17 : Vérification que les valeurs reçues ne sont pas vides.
Lignes 19 à 22 : Écriture sur la socket ZeroMQ PUSH des données reçues.

Code du contrôleur Event

Ce contrôleur a pour rôle d’envoyer par SSE les événements reçus sur sa socket ZeroMQ.

Voici le contenu du fichier controller/Event.php :

<?php

/** Contrôleur Event. */
class Event extends \Temma\Web\EventsController {
    /** Action qui envoie les notifications SSE. */
    public function fetch() {
        // boucle infinie
        while (true) {
            // lecture d'un message envoyé par le broker
            $message = $this->zmq_client_sub[''];
            // envoi du message au client sur le canal "msg"
            $this['msg'] = $message;
        }
    }
}
Langage du code : PHP (php)

Ligne 4 : Définition du contrôleur Event, qui est un contrôleur d’événements (il est fait pour envoyer des SSE, et non pas pour renvoyer un contenu transactionnel comme du HTML ou du JSON).
Ligne 6 : définition de l’action fetch.
Ligne 8 : boucle infinie.
Ligne 10 : récupération d’un message envoyé par le broker, en lisant sur la socket ZeroMQ SUB.
Ligne 12 : envoi du message reçu au client par SSE.

Code de la page web

La page web qui est chargée sur les navigateurs comporte deux parties : une liste dans laquelle vont s’ajouter les messages, et un formulaire qui va servir à envoyer de nouveaux messages.

Voici le contenu du fichier www/minichat.html :

<!DOCTYPE html>
<html>
<body>
    <!-- liste des messages -->
    <ul id="liste"></ul>
    <!-- formulaire -->
    <input id="from" type="text" placeholder="Nom" autofocus>
    <input id="text" type="text" placeholder="Message">
    <button onclick="sendMessage()">Envoyer</button>
    <!-- chargement du code javascript -->
    <script src="/minichat.js"></script>
</body>
</html>Langage du code : HTML, XML (xml)

Code Javascript

Et voici le code Javascript chargé par la page, dans le fichier www/minichat.js :

// connexion à la source SSE
const evtSource = new EventSource("/event/fetch");
// écoute des événements envoyés sur le canal "msg"
evtSource.addEventListener("msg", function(event) {
    // création de l'élément DOM <li>
    const newElement = document.createElement("li");
    // désérialisation de l'événement reçu
    var evtData = JSON.parse(event.data);
    // ajout du message dans la liste
    newElement.textContent = evtData.from + " : " + evtData.text;
    // ajout de l'élément à la liste dans la page
    document.getElementById("liste").appendChild(newElement);
});

// fonction appelée pour envoyer les messages au serveur
function sendMessage() {
    // récupération des valeurs du formulaire
    const from = document.getElementById("from").value;
    const text = document.getElementById("text").value;
    // création de l'objet de paramètres POST
    const formData = new FormData();
    formData.append("from", from);
    formData.append("text", text);
    // envoi des données en AJAX
    var xhr = new XMLHttpRequest();
    xhr.open("POST", "/message/broadcast", true);
    xhr.send(formData);
}Langage du code : JavaScript (javascript)

Conclusion

Cet exemple n’est qu’un démonstrateur technique. Plusieurs améliorations sont facilement imaginables :

Enregistrer les messages en base de données, pour pouvoir afficher un historique.
Utiliser une file de messages, comme Beanstalkd ou AWS SQS. Ce n’est pas si évident, car le code serait plus complexe (on est ici sur du broadcasting, pas sur du traitement de tâches par des workers). ZeroMQ est une solution très efficace dans un cas comme celui-ci.
Faire évoluer le broker pour qu’il gère la persistance des messages (sur disque ou en base de données), au cas où le broker − ou le serveur − s’arrête alors que des messages n’ont pas encore été traités.
Mettre en place un système de supervision du broker, pour s’assurer qu’il tourne en permanence (comme Supervisor, God ou Monit).

Mais sur le principe, tout y est, avec peu de lignes de code.

De geek à directeur technique