Temps réel : des flux vers le navigateur

01Comment une diffusion atteint une page

Une page s’abonne à un ou plusieurs flux nommés ; n’importe quelle partie de l’application diffuse vers un flux par son nom ; chaque page abonnée reçoit le message en un dixième de seconde environ.

• Les flux ont des noms. Un nom est un texte court quelconque ("alerts"), ou un enregistrement, dont le nom est sa table et son id ("articles:12", Cybertrain::Stream.name_for(@article)).
• Une diffusion est une ligne. Cybertrain::Stream.broadcast(name, payload) insère une ligne dans cybertrain_messages, une table que le framework crée lui-même dans la base de données de l’application. Dans une DB.transaction, le message ne part que si la transaction est validée. Parce que c’est une ligne, une diffusion fonctionne depuis tout processus qui partage la base : une requête, un job dans un processus worker, un callback de flux, une Main.command.
• Chaque processus web interroge la table. Un thread par processus server (le poller des flux) lit les nouvelles lignes toutes les stream_poll_ms (100 ms) et remet chacune aux navigateurs que ce processus tient. Il supprime une fois par minute les lignes plus vieilles que stream_retention (60 secondes).
• Le navigateur se connecte à un seul point d’entrée, /cybertrain/stream, en Server-Sent Events (EventSource) ou en WebSocket, avec un jeton signé qui nomme les flux qu’il a le droit de lire (Noms de flux signés).

Il n’y a rien à démarrer : ./<app> server fait tourner le poller et sert le point d’entrée. config.stream_path = "" désactive toute la fonctionnalité.

02Turbo Streams

Turbo met une page à jour à partir d’éléments <turbo-stream action="append" target="comments">, comme le font les applications Rails 8. Chargez Turbo dans le layout (il n’y a pas d’asset pipeline : un CDN ou une copie dans public/), et abonnez une page avec turbo_stream_from :

<head>
  <%= csrf_meta_tags %>
  <script type="module" src="https://cdn.jsdelivr.net/npm/@hotwired/turbo@8/dist/turbo.es2017-esm.js"></script>
</head>

<%= turbo_stream_from @article %>
<div id="comments">
  <% @article.comments.each do |comment| %>
    <%= render "comments/comment", comment: comment %>
  <% end %>
</div>

turbo_stream_from prend des Strings et des enregistrements, autant que voulu, et écrit un seul élément <turbo-stream-source> dont l’URL porte un seul jeton pour tous. Turbo l’ouvre comme un EventSource ; turbo_stream_from @article, ws: true donne à la place une URL ws:// (wss:// derrière HTTPS), que Turbo ouvre comme un WebSocket. Dans les deux cas, la page n’a besoin d’aucun JavaScript de votre part. Turbo ne rouvre pas un WebSocket perdu (coupure réseau, redémarrage du serveur) : cette page ne reçoit plus rien jusqu’à son rechargement, là où un EventSource se reconnecte tout seul et rejoue ce qu’il a manqué. Gardez la valeur par défaut sauf besoin de WebSocket, ou reconnectez depuis votre propre JavaScript (&after=, ci-dessous).

Diffusez ensuite depuis l’action qui a changé quelque chose :

def create
  @comment = Comment.new(comment_params)
  @comment.article_id = @article.id
  if @comment.save
    Cybertrain::Stream.broadcast_append_to(@article, target: "comments",
                                           partial: "comments/comment", locals: { comment: @comment })
  end
  redirect_to article_path(@article), status: :see_other
end

broadcast_append_to(stream, target:, ...) rend le contenu, l’enveloppe dans un élément <turbo-stream> et le diffuse. Le contenu est l’un de partial: avec locals: (rendu par Cybertrain::Views.render_partial), component: (un composant de vue) ou html: (échappé sauf s’il s’agit d’une SafeString). La cible est un id DOM, ou un enregistrement, dont l’id DOM est dom_id(record) ("comment_7"). Les autres fonctionnent de la même façon : broadcast_replace_to, broadcast_update_to, broadcast_prepend_to, broadcast_before_to, broadcast_after_to, broadcast_remove_to(stream, target:) et broadcast_refresh_to(stream) (Turbo 8 recharge la page avec le morphing).

Turbo transforme aussi les liens et les formulaires en requêtes fetch (Turbo Drive). Le scaffold répond déjà comme Turbo l’attend : une redirection en 303 après un formulaire réussi, et 422 avec le formulaire à nouveau après un échec.

Répondre à un formulaire par un Turbo Stream

Un formulaire que Turbo soumet accepte text/vnd.turbo-stream.html. Pour répondre par un flux plutôt que par une redirection, rendez l’élément vous-même ; Cybertrain::Turbo les construit (replace, update, append, prepend, before, after, remove, refresh) :

def destroy
  @comment = Comment.where(article_id: @article.id).find(params[:id])
  @comment.destroy
  if request.header("accept").to_s.include?("text/vnd.turbo-stream.html")
    render html: Cybertrain::Turbo.remove(@comment), content_type: "text/vnd.turbo-stream.html"
  else
    redirect_to article_path(@article), status: :see_other
  end
end

Diffuser depuis un job, une tâche ou un flux

Une diffusion n’a besoin d’aucune requête, donc un job peut en rendre et en envoyer une :

Cybertrain::Job.handle("price_refresh") do |payload|
  item = Item.find(Cybertrain::Cast.int(payload))
  item.refresh_price!
  Cybertrain::Stream.broadcast_replace_to("prices", target: item,
                                          partial: "items/price", locals: { item: item })
end

Le partial est rendu par Cybertrain::Views.render_partial, qui rend sans requête : les helpers de chemin (item_path(item)) fonctionnent, et les helpers qui ont besoin d’une requête (csrf_token, params, flash, request_path, *_url) lèvent Cybertrain::Views::OutsideRequest (voir Rendre hors d’une requête). Un job tourne dans un processus worker aussi bien que dans un serveur, et la ligne qu’il écrit atteint les pages de tous les serveurs.

03Server-Sent Events depuis JavaScript

Pour une page que Turbo ne pilote pas (un graphique, un téléscripteur), demandez les flux en JSON. stream_source_path donne le chemin du point d’entrée avec le jeton et mode=json, et signed_stream_name le jeton seul :

<div id="price" data-src="<%= stream_source_path "prices", "alerts" %>"></div>
<script type="module">
  const el = document.getElementById("price");
  const source = new EventSource(el.dataset.src);
  source.addEventListener("prices", (event) => {
    const message = JSON.parse(event.data);   // {"stream":"prices","id":42,"data":"..."}
    el.textContent = JSON.parse(message.data).price;
  });
  source.addEventListener("alerts", (event) => console.log(JSON.parse(event.data).data));
</script>

En mode=json, chaque message est un événement nommé d’après son flux, dont les données sont {"stream": ..., "id": ..., "data": ...} avec la charge utile dans data (diffusez du JSON sous forme de String : Stream.broadcast("prices", JSON.generate({ "price" => 12.5 }))). Dans le mode=raw par défaut, la charge utile part telle quelle, comme un simple événement message, ce que lit Turbo.

Le navigateur se reconnecte tout seul quand la connexion tombe, et envoie l’id du dernier événement qu’il a vu (Last-Event-ID). Le point d’entrée rejoue alors d’abord chaque message de ces flux dont l’id est plus grand et qui est encore dans la table (dans la limite de stream_retention), si bien que rien n’est perdu entre-temps. Une page absente plus longtemps devrait se recharger. Avant l’arrivée de son premier événement, le navigateur n’a pas d’id à envoyer : une connexion qui tombe dans cette fenêtre ne rejoue rien, et ce que la page a rendu au chargement reste son état jusqu’à la diffusion suivante. Une ligne de commentaire (: ping) part après 15 secondes de silence pour que les proxies gardent la connexion ouverte.

04WebSocket depuis JavaScript

La même URL répond à un WebSocket. Une page en tire une URL ws:// ou wss://. Passez le chemin et les jetons par des attributs data-, comme ici : un <%= %> à l’intérieur d’un élément <script> est échappé en HTML, et le navigateur ne le décode pas à cet endroit (& atteindrait le serveur).

<div id="ticker" data-src="<%= stream_source_path "prices" %>" data-alerts="<%= signed_stream_name "alerts" %>"></div>
<script type="module">
  const el = document.getElementById("ticker");
  const scheme = location.protocol === "https:" ? "wss://" : "ws://";
  const socket = new WebSocket(scheme + location.host + el.dataset.src);
  socket.addEventListener("message", (event) => {
    const message = JSON.parse(event.data);
    if (message.stream) console.log(message.stream, message.id, message.data);
  });
  socket.addEventListener("open", () => {
    socket.send(JSON.stringify({ subscribe: el.dataset.alerts }));
  });
</script>

Avec mode=json, chaque message est une trame texte qui contient la même enveloppe {"stream", "id", "data"}, et la page peut changer de flux en cours de connexion en envoyant une trame texte JSON :

Une connexion en mode=json peut aussi s’ouvrir sans jeton et s’abonner par message ; celle qui ne lit toujours aucun flux au bout de 15 secondes est fermée (code 1008). Un WebSocket n’a pas de Last-Event-ID : pour rattraper son retard après une reconnexion, ajoutez &after=<the last id you saw> à l’URL. Le serveur envoie un ping après chaque silence de 15 secondes et répond aux pings ; les trames binaires de la page sont ignorées. Codes de fermeture : 1000 quand le serveur s’arrête, 1008 pour un mauvais jeton, 1009 pour une trame au-delà de stream_max_message, 1013 quand le processus est plein ou que la page a pris trop de retard.

05Diffuser en continu depuis une action

Une action peut garder sa propre réponse ouverte, comme le fait ActionController::Live de Rails : stream exécute le bloc après l’action, sur le thread de la connexion, et lui donne le socket sous la forme d’un Cybertrain::Sink.

def progress
  stream do |out|
    10.times do |i|
      out.write(Cybertrain::SSE.event(JSON.generate({ "done" => i * 10 }), event: "progress", id: i.to_s))
      break if out.wait_closed(1)   # vrai quand le client est parti
    end
  end
end

stream(content_type: "text/event-stream", status: :ok) répond avec Transfer-Encoding: chunked et ferme la connexion quand le bloc rend la main. SSE.event(data, event: "", id: "", retry_ms: 0) met en forme un événement ; out.closed? devient vrai une fois le client parti, et les écritures suivantes ne font rien. Le bloc ne doit toucher ni à la session, ni au flash, ni aux en-têtes : ils sont partis avec l’en-tête de la réponse.

06Noms de flux signés

Une page ne peut lire un flux qu’avec un jeton qui le nomme. Le jeton est formé des noms de flux et d’une signature HMAC-SHA256 faite avec une clé dérivée de secret_key_base pour les seuls flux (comme turbo/signed_stream_verifier_key dans Rails), si bien qu’un jeton n’est jamais un cookie de session valide ; le point d’entrée vérifie la signature et ne lit ni la session ni la base de données. C’est le modèle de Turbo : qui pouvait rendre la page pouvait lire ses flux. Les jetons n’expirent pas ; changer secret_key_base les invalide tous, et une page restée ouverte à travers ce déploiement ne reçoit plus rien (un EventSource tient le 403 pour définitif) jusqu’à son rechargement.

• Un flux propre à un utilisateur est un flux nommé d’après l’enregistrement de cet utilisateur, turbo_stream_from @user ("users:7"), rendu seulement sur les pages de cet utilisateur. Quiconque obtient le jeton peut lire ce flux, tant que le secret ne change pas.
• Les jetons n’expirent pas. Changer CYBERTRAIN_SECRET_KEY_BASE invalide tous les jetons (et toutes les sessions).
• Le jeton voyage dans l’URL, donc le journal d’accès d’un reverse proxy l’enregistre. Le journal des requêtes du framework l’écrit [FILTERED] ; désactivez la journalisation du proxy pour /cybertrain/stream, ou gardez ces journaux aussi privés que les flux.
• Un jeton invalide ou absent donne un 403 en SSE, et une fermeture en 1008 juste après la poignée de main en WebSocket.
• Un nom fait de 1 à 200 octets, sans caractère de contrôle ni virgule ; un jeton porte au plus 16 noms, une connexion au plus 64.

07Limites et réglages

Réglez-les dans config/app.rb (c.stream_poll_ms = 250) ; le démarrage s’arrête avec un message pour une valeur hors limites.

Chaque flux ouvert occupe une connexion, et son green thread, tant que la page est ouverte. Derrière nginx, l’en-tête X-Accel-Buffering: no de la réponse désactive la mise en tampon ; pour WebSocket, le proxy doit transmettre Upgrade et Connection (Caddy et kamal-proxy le font).

08Quel processus fait quoi

• ./<app> server fait tourner le poller et le point d’entrée, toujours, même avec CYBERTRAIN_WORKER=0 : il livre aux navigateurs qu’il tient. Tant que le poller ne tourne pas, le point d’entrée répond 503 (stream poller is not running) ; serve le démarre avant d’ouvrir le port, si bien qu’un processus servi ne le montre jamais.
• ./<app> worker et ./<app> feeds ne servent pas de HTTP et ne font tourner aucun poller. Leurs diffusions sont des lignes, que les serveurs livrent (Rôles des processus).
• Plusieurs serveurs sur une même base livrent chacun chaque message à leurs propres navigateurs, dans le même ordre et avec les mêmes ids.

Tester le flux d’une page : Tester les flux et les composants. La conception : docs/design.md D36 et D37.