Préface

Avant d'aller plus loin, expliquons d'abord ce que sont les Server-sent Events (SSE pour les intimes).

Il s'agit d'un canal unidirectionnel permettant au serveur d'envoyer des messages à un navigateur.

C'est une techno ancienne, qui date de 2009-2010 et est disponible au grand public depuis les versions 6 de Chrome et Firefox, mais qui a été un peu mise de coté à l'arrivée des WebSockets.

Objectif ?

Le but de cette démo est de construire un cas concret d'utilisation de SSE.
Beaucoup d'articles ou de démos que j'ai pu trouver ne montrent guère autre chose qu'un event pré-construit envoyé à intervalle régulier, pas grand-chose à voir avec la réalité.

Dans cette démo, nous allons construire une API Rest, qui poussera un message dans un RabbitMQ en mode Fan-out.
De l'autre coté, nous aurons un serveur nodejs qui écoutera les messages et poussera un SSE au bon client de la webapp servi par le nodejs.

Dans l'objet poussé sur l'endpoint, nous avons un client ID, ce qui permet d'avoir plusieurs onglets ouverts sur la web app et ainsi voir que les SSE arrivent seulement au bon client.

Le code :

L'API Rest

Je ne vais pas entrer ici dans les détails de l'API Rest, c'est une API Java (Jooby 2) qui sert de producer RabbitMQ.

La Web App

Il s'agit d'une simple page html avec un Web Component (les dépendances sont chargées directement avec Pika, cela peut prendre quelques secondes).

Pour se connecter à un endpoint SSE, rien de plus simple, il suffit de définir un EventSource qui prend en paramètre l'url de la source.

const sseSource = new EventSource("/event-stream/" + this.clientId);

Dans notre cas, nous ajoutons dans l'url le clientId du client connecté.

Sur cet objet EventSource nous allons venir définir un event listener, le but est de ne réagir qu'aux messages de type notif :

    sseSource.addEventListener("notif", e => {
      this.count = this.count + 1;
      this.notifications = [...this.notifications, e.data];
    });

Ce code nous permet d'incrémenter un compteur et d’alimenter un tableau avec le message de l’événement.
C'est tout pour la partie front.

La partie serveur NodeJs

1) Le point d'entrée SSE

Ici on utilise ExpressJS car il n'y rien de particulier à faire pour mettre en place la ressource SSE :

app.get("/event-stream/:clientId", (req, res) => {
  let clientId = req.params.clientId;

  res.writeHead(200, {
    "Content-Type": "text/event-stream",
    "Cache-Control": "no-cache",
    Connection: "keep-alive"
  });
  res.write("\n");

  connections[clientId] = res;

  console.log("Connection open for client : " + clientId);

  req.on("close", () => {
    delete connections[clientId];
  });
});

Ici nous définissons notre ressource GET "/event-stream/:clientId" avec un paramètre clientId qui nous permet de stocker l'objet response sur lequel envoyer les messages :

    connections[clientId] = res;

Mais avant de stocker cette réponse, il faut la "configurer" pour le SSE :

  res.writeHead(200, {
    "Content-Type": "text/event-stream",
    "Cache-Control": "no-cache",
    Connection: "keep-alive"
  });

Pour résumer cette partie, à chaque fois qu'un client fait un GET sur notre ressource SSE, nous stockons dans un objet sont clientId et l'objet response afin de lui pousser des messages ultérieurement.

2) Le consumer RabbitMQ

Afin de pouvoir se connecter sur le RabbitMQ, nous allons avoir besoin de la dépendance vers la librairie amqplib

   const amqp = require("amqplib/callback_api"); // => fonctionne avec des callback
// ou
   const amqp = require("amqplib");  // => fonctionne avec des promesses

Au démarrage du serveur nodejs, on ouvre une connexion vers RabbitMQ

amqp.connect("amqp://localhost", (error0, connection) => {})

Dans cette connexion, on va créer un channel sur notre Exchange et une Queue (Cf dépôt Github pour le code de cette partie).

Ce qui nous intéresse ici, c'est de voir comment on consomme les messages.

      channel.consume(
        q.queue,
        msg => {
          if (msg.content) {
            let data = JSON.parse(msg.content.toString());

            console.log("New message for client : " + data.clientId);
            let clientRes = connections[data.clientId];

            if (clientRes) {
              clientRes.write(`event: notif\n`);
              clientRes.write(`data: ${data.content}\n\n`);
            }
          }
        },
        {
          noAck: true
        }
      );

On va parser le message pour en récupérer le contenu, grâce au clientId du message, on récupère notre objet response qui correspond.
Si elle existe, on vient écrire une première ligne avec le type d'événement clientRes.write('event: notif\n'); (pour rappel notre front n'écoute que les messages notif).
Puis une seconde ligne avec le contenu du message clientRes.write('data: ${data.content}\n\n');

Et voilà, nous avons poussé notre message dans notre stream d'événements.
Le front réagi, incrémente le compteur et affiche le message !

J'espère que maintenant l'utilisation des SSE est un peu plus claire et concrète pour vous aussi :)

Le code de la démo complète est disponible sur GitHub :