Hub en temps réel (WebSockets et SSE)

bext est livré avec un hub de pub/sub en temps réel intégré qui prend en charge les transports SSE (Server-Sent Events) et WebSocket. Aucun courtier de messages externe n'est requis pour les déploiements à instance unique : le hub s'exécute en cours de processus sans infrastructure supplémentaire. Pour la mise à l'échelle horizontale, un relais Redis en option synchronise les événements entre les instances. Il s'agit d'une fonctionnalité Pro.

Configuration#

Activez et configurez le hub dans bext.config.toml :

toml

[realtime]
enabled = true
max_connections = 10000       # 0 = unlimited
heartbeat_interval_ms = 30000 # ping/keepalive interval
replay_buffer_size = 1000     # events kept for Last-Event-ID catchup

# Optional: Redis relay for multi-instance sync
[realtime.redis]
url = "rediss://redis.internal:6379"
prefix = "bext:"              # channel becomes "bext:hub:events"

Caractères génériques de sujet#

Les sujets utilisent des séparateurs de chemin de style MQTT ( / ) avec deux types de caractères génériques :

Modèle	Matchs	Ne correspond PAS
`app/deploy`	`app/deploy` (exact)	`app/restart`
`app/*`	`app/deploy`, `app/restart`	`app/deploy/us-east`
`app/#`	`app/deploy`, `app/deploy/us-east`, `app/x/y/z`	`system/deploy`

* correspond exactement à un segment de chemin. # correspond à zéro ou plusieurs segments de fin et doit être le dernier segment du modèle.

Protocole WebSocket#

Connectez-vous à wss://your-domain/__bext/ws et échangez des messages au format JSON.

Messages client-serveur :

json

// Subscribe to topics (supports wildcards)
{"type": "subscribe", "topics": ["chat/room-42", "notifications/#"]}

// Unsubscribe from specific topics
{"type": "unsubscribe", "topics": ["chat/room-42"]}

// Publish to a topic
{"type": "publish", "topic": "chat/room-42", "data": {"text": "hello"}}

// Respond to server heartbeat
{"type": "pong"}

Messages serveur-client :

json

// Subscription confirmed
{"type": "subscribed", "topics": ["chat/room-42", "notifications/#"]}

// Event delivery
{"type": "event", "topic": "chat/room-42", "data": {"text": "hello"}, "id": 571}

// Heartbeat (client must reply with pong)
{"type": "ping"}

// Error (e.g. max connections reached, auth failure)
{"type": "error", "message": "max connections reached"}

Exemple client ( JavaScript )#

const ws = new WebSocket("wss://app.example.com/__bext/ws");

ws.onopen = () => {
  ws.send(JSON.stringify({
    type: "subscribe",
    topics: ["orders/#", "notifications/*"]
  }));
};

ws.onmessage = (e) => {
  const msg = JSON.parse(e.data);
  if (msg.type === "event") {
    console.log(`[${msg.topic}]`, msg.data);
  } else if (msg.type === "ping") {
    ws.send(JSON.stringify({ type: "pong" }));
  }
};

Événements envoyés par le serveur (SSE)#

Pour les abonnés en lecture seule (tableaux de bord, flux en direct), SSE est plus simple que WebSockets :

code

GET /__bext/events?topics=orders/#,notifications/*
Accept: text/event-stream

La réponse est un flux SSE standard :

code

event: orders/new
data: {"id":"ord_839","total":59.99}
id: 571

event: notifications/system
data: {"message":"deploy complete"}
id: 572

Utilisez l'en-tête Last-Event-ID lors de la reconnexion pour relire les événements manqués à partir du tampon en anneau :

const source = new EventSource("/__bext/events?topics=orders/#");
source.onmessage = (e) => console.log(e.data);
// Browser automatically sends Last-Event-ID on reconnect

Crochets d'authentification#

Le hub prend en charge l'autorisation au niveau du sujet avec quatre types de stratégie :

Politique	Description
`public`	Tout le monde, y compris les utilisateurs anonymes
`authenticated`	Tout utilisateur disposant d'une session/JWT valide
`role`	Uniquement les utilisateurs avec un rôle spécifique (par exemple `admin` )
`user_id`	Uniquement un utilisateur spécifique par ID

Les stratégies sont configurées par modèle de sujet, avec des règles distinctes pour l'abonnement et la publication :

toml

[[realtime.auth_rules]]
pattern = "system/#"
subscribe_policy = { type = "authenticated" }
publish_policy = { type = "role", value = "__system_internal__" }

[[realtime.auth_rules]]
pattern = "user/*/inbox"
subscribe_policy = { type = "user_id", value = "$user_id" }
publish_policy = { type = "authenticated" }

[[realtime.auth_rules]]
pattern = "broadcast/#"
subscribe_policy = { type = "public" }
publish_policy = { type = "role", value = "broadcaster" }

La première règle correspondante gagne. Les sujets sans correspondance bénéficient des valeurs par défaut : abonnement public, publication authentifiée.

Cycle de vie de la connexion#

Connecter : le client ouvre une connexion WebSocket ou SSE.
Abonnez-vous -- le client envoie des modèles de sujet ; le hub enregistre un abonné avec un canal limité à 256 événements.
Heartbeat : le serveur envoie des trames ping à l'intervalle configuré. Les clients qui ne répondent pas dans le délai d'attente pong (10 s par défaut) sont déconnectés.
Expulsion lente du client : si le canal d'un abonné est saturé (256 événements en attente), le hub interrompt la connexion pour empêcher une croissance illimitée de la mémoire.
Déconnexion -- à la fermeture de WebSocket ou à une interruption de la connexion SSE, l'abonné est automatiquement désenregistré et toutes les associations de sujets sont nettoyées.

Relais Redis pour synchronisation multi-instance#

Lors de l'exécution de plusieurs instances bext derrière un équilibreur de charge, les événements publiés sur une instance doivent atteindre les abonnés sur toutes les instances. Le relais Redis relie les événements du hub local à un canal Pub/Sub Redis partagé.

Chaque instance génère un identifiant unique au démarrage (par exemple bext-12345-789012 ) et balise les messages sortants. Les messages entrants provenant du même ID d'instance sont supprimés silencieusement pour éviter les boucles d'écho. Le relais se reconnecte automatiquement avec un intervalle exponentiel (100 ms à 30 s) en cas d'échec de connexion Redis.

toml

[realtime.redis]
url = "rediss://redis.internal:6379"  # Use rediss:// for TLS
prefix = "myapp:"

Surveillance#

Le hub expose des statistiques en direct sur /__bext/admin/api/realtime :

json

{
  "active_connections": 342,
  "total_published": 18492,
  "total_delivered": 847201,
  "topic_count": 27,
  "subscriber_count": 342,
  "uptime_secs": 86412.5
}

Ces métriques sont également disponibles au format Prometheus à l'adresse /__bext/metrics.

Attention

Les connexions SSE sur les routes /api/* tamponnent jusqu'à la fin plutôt que de diffuser en direct. Utilisez le point de terminaison SSE /__bext/events intégré (décrit ci-dessus) pour la livraison en temps réel. Consultez le Note de diffusion d'itinéraires API bext si vous avez besoin de SSE à partir d’une route API personnalisée.

En rapport#

Protocoles : ESS — Détails du protocole des événements envoyés par le serveur et sémantique de relecture Last-Event-ID
Protocoles : MQTT — transport pub/sub alternatif avec la même syntaxe de caractère générique de sujet
Mise en cache — ISR et SWR ; les points de terminaison en temps réel doivent définir cache = false dans leur règle de routage
Compression — Les routes SSE doivent désactiver la compression ( compression = "off" ) pour éviter la mise en mémoire tampon
Capacités : Authentification — validation de session et JWT utilisée par les hooks d'authentification du hub

Links#

MDN : API WebSockets — référence WebSocket côté client
MDN : événements envoyés par le serveur — API EventSource et format de fil SSE