Server-Sent Events (SSE) est une technologie permettant la communication unidirectionnelle en temps réel du serveur vers le client via une connexion HTTP persistante. Elle est particulièrement adaptée aux scénarios où le serveur doit pousser des mises à jour fréquentes vers le client, comme les notifications, les flux d'actualités ou les mises à jour dynamiques de contenu. Comparé à WebSocket, SSE est plus léger et plus simple à implémenter pour des flux de données unidirectionnels.
Principes Fondamentaux de SSE
SSE repose sur une connexion HTTP de longue durée et utilise le concept de ReadableStream pour un flux de données du serveur vers le client. Le navigateur fournit un objet EventSource intégré pour gérer facilement ces flux, tandis que le serveur peut envoyer des événements en écrivant continuellement sur l'objet Response. Dans des cas d'utilisation réels, plusieurs défis peuvent survenir :
- Pré-traitement des Réponses côté Serveur : Lors de la construction d'applications conversationnelles ou de flux de données complexes, il est souvent nécessaire de traiter les données reçues du modèle d'IA (par exemple, filtrage, censure) avant de les transmettre au client. Cela implique d'intercepter le flux de données côté serveur, de le traiter, puis de le renvoyer en flux.
- Transfert Direct des Données Serveur : Si aucun pré-traitement n'est requis, le transfert direct de la réponse du serveur cible vers le client peut être plus efficace que de la recevoir puis de la réexpédier. Cela peut être réalisé en agsisant comme un proxy HTTP pour la connexion longue durée.
- Utilisation de
fetchpour les Requêtes : L'objetEventSourcestandard ne prend en charge que les requêtes GET et ne permet pas de définir des en-têtes personnalisés ou d'envoyer un corps de requête. Pour des besoins d'authentification ou des configurations plus complexes, l'utilisation defetchdevient nécessaire, car il offre une flexibilité totale sur les requêtes HTTP.
Nous commencerons par explorer l'implémentation de base de SSE en utilisant l'objet EventSource côté client. Pour le serveur, nous utiliserons Node.js pour simuler la réponse en flux.
Réponse en Flux côté Serveur avec Node.js
Pour implémenter une réponse en flux, le serveur doit définir l'en-tête Content-Type sur text/event-stream. Il est crucial que cet en-tête soit envoyé avant tout corps de réponse pour éviter des erreurs d'encodage.
// packages/fetch-sse/server/modules/ping.ts
import http from 'http';
const ping = (req: http.IncomingMessage, res: http.ServerResponse<http.incomingmessage>) => {
res.writeHead(200, {
"Content-Type": "text/event-stream; charset=utf-8",
"Cache-Control": "no-cache",
"Connection": "keep-alive",
});
// Le reste du code pour envoyer les données...
};
</http.incomingmessage>
Le protocole SSE définit un format spécifique pour les événements. Chaque événement est délimité par deux nouvelles lignes (\n\n). Les champs à l'intérieur d'un événement (comme id, event, et data) sont séparés par une seule nouvelle ligne (\n).
id: 1
event: message
data: hello world
id: 2
event: custom
data: hello
data: world
L'objet EventSource du navigateur gère automatiquement la reconnexion et la gestion des identifiants d'événements. Pour une implémentation personnalisée avec fetch, ces fonctionnalités devront être gérées manuellement. Nous pouvons également spécifier des événements personnalisés côté serveur.
// packages/fetch-sse/server/modules/ping.ts
// ... (en-têtes déjà définis)
res.write("retry: 10000\n"); // Délai de reconnexion en ms
res.write("id: -1\n"); // ID de l'événement
res.write("event: connect\n"); // Nom de l'événement
res.write("data: " + new Date() + "\n\n"); // Données de l'événement
Côté client, l'objet EventSource peut écouter des événements spécifiques ou l'événement par défaut message.
// packages/fetch-sse/client/components/ping.tsx
import { useMemoFn } from 'your-hook-library'; // Assurez-vous que cette bibliothèque est disponible
const prepend = (text: string) => {
// Logique pour ajouter du texte à l'élément DOM
const el = ref.current; // Assumant que ref est défini ailleurs
if (!el) return;
const child = document.createElement("div");
child.textContent = text;
el.prepend(child);
};
const onConnect = useMemoFn((e: MessageEvent<string>) => {
prepend("Start Time: " + e.data);
});
const source = new EventSource("/ping");
source.addEventListener("connect", onConnect);
</string>
Pour envoyer des mises à jour continues, le serveur peut utiliser un minuteur pour envoyer périodiquement des données.
// packages/fetch-sse/server/modules/ping.ts
let index = 0;
const interval = setInterval(() => {
res.write(`id: ${index++}\n`);
res.write(`data: ${new Date()}\n\n`);
}, 1000);
// Gestion de la fermeture de la connexion côté client
req.socket.on("close", () => {
console.log("[ping] connection close");
clearInterval(interval);
res.end();
});
Côté client, l'écoute de l'événement message se fait via onmessage ou addEventListener("message", ...).
// packages/fetch-sse/client/components/ping.tsx
const onMessage = (e: MessageEvent<string>) => {
prepend("Ping: " + e.data);
};
// ... (création de source)
source.onmessage = onMessage;
</string>
Il est essentiel de fermer la connexion côté serveur lorsque le client se déconnecte pour éviter les fuites de ressources.
Limites de Connexion SSE
SSE est soumis à une limite de connexions simultanées par domaine (généralement 6 par navigateur, sauf en HTTP/2). HTTP/2 augmente considérablement cette limite, car les requêtes sont multiplexées sur une seule connexion.
Implémentation Serveur : Pré-traitement et Transfert
Simulation de Flux de Données
Pour simuler des flux de données dynamiques, nous pouvons créer une route /stream qui renvoie des données progressivement. Comme le caractère \n est utilisé dans le protocole SSE, il doit être échappé (par exemple, remplacé par \\n) s'il fait partie des données, notamment lors du traitement de Markdown.
// packages/fetch-sse/server/modules/stream.ts
const content = `# Extrait du "Memorial d'Uzushiogakure"\n\n- Par Zhuge Liang\n\n... (contenu du texte)`; // Texte simulé
// ... (Configuration des en-têtes response)
res.write("event: connect\n");
res.write("data: " + Date.now() + "\n\n");
let start = 0;
const interval = setInterval(() => {
const sliceEnd = Math.min(start + Math.floor(Math.random() * 30) + 1, content.length);
res.write("event: message\n");
// Remplacer les nouvelles lignes dans les données pour éviter les interférences avec le protocole SSE
res.write("data: " + content.slice(0, sliceEnd).replace(/\n/g, "\\n") + "\n\n");
start = sliceEnd;
if (start >= content.length) {
clearInterval(interval);
res.end();
}
}, 500);
req.socket.on("close", () => {
console.log("[stream] connection close");
clearInterval(interval);
res.end();
});
Transfert de Données SSE
Pour implémenter le transfert de données SSE, le serveur agit comme un proxy. Il effectue une requête vers la source de données SSE et relaie les événements au client connecté. La bibliothèque node-fetch est utilisée pour effectuer la requête initiale.
// packages/fetch-sse/server/modules/transfer.ts
import fetch from "node-fetch";
import { AbortController } from 'node-abort-controller'; // Pour la gestion de l'annulation
const transfer = async (req: http.IncomingMessage, res: http.ServerResponse) => {
const targetUrl = "http://127.0.0.1:8800/stream";
const ctrl = new AbortController();
req.socket.on("close", () => {
console.log("[transfer] connection close");
ctrl.abort();
res.end();
});
try {
const response = await fetch(targetUrl, { signal: ctrl.signal as AbortSignal });
const readable = response.body;
if (!readable) {
res.writeHead(500, { "Content-Type": "text/plain" });
res.end("Failed to get stream body");
return;
}
// Configuration des en-têtes de réponse pour le client
res.writeHead(response.status, {
"Content-Type": "text/event-stream",
// Copier d'autres en-têtes pertinents si nécessaire
});
// Traitement du flux de données SSE entrant et renvoi au client
const parser = new StreamParser(); // Classe personnalisée pour parser SSE
parser.onMessage = (message) => {
res.write(`event: ${message.event}\n`);
res.write(`data: ${message.data}\n\n`);
};
for await (const chunk of readable) {
parser.onBinary(chunk as Uint8Array); // Assumant que chunk est Uint8Array ou Buffer
}
res.end();
} catch (error) {
if (error.name === 'AbortError') {
console.log('Transfer aborted');
} else {
console.error("Transfer error:", error);
res.writeHead(502, { "Content-Type": "text/plain" });
res.end("Bad Gateway");
}
}
};
// Classe StreamParser (simplifiée pour l'exemple)
class StreamParser {
private buffer = new Uint8Array(0);
private message: any = {};
public onMessage: (message: any) => void = () => {};
private compose(data: Uint8Array) {
const buffer = new Uint8Array(this.buffer.length + data.length);
buffer.set(this.buffer);
buffer.set(data, this.buffer.length);
this.buffer = buffer;
return buffer;
}
private onLine(bytes: Uint8Array) {
if (bytes.length === 0) {
if (this.onMessage && this.message.event) {
this.message.data = this.message.data || "";
this.onMessage(this.message);
}
this.message = {};
return;
}
const decoder = new TextDecoder();
const line = decoder.decode(bytes);
const [field, ...rest] = line.split(":");
const value = rest.join(":").trim();
switch (field) {
case "id": this.message.id = value; break;
case "event": this.message.event = value; break;
case "data":
this.message.event = this.message.event || "message";
this.message.data = value;
break;
}
}
public onBinary(bytes: Uint8Array) {
const buffer = this.compose(bytes);
let start = 0;
for (let i = 0; i < buffer.length; i++) {
if (buffer[i] === 10) { // ASCII pour \n
this.onLine(buffer.slice(start, i));
start = i + 1;
}
}
this.buffer = buffer.slice(start);
}
}
Proxy HTTP Direct
Lorsque aucun pré-traitement n'est nécessaire, le serveur peut directement faire office de proxy. Il établit une connexion HTTP longue durée vers l'URL cible et relaie les données au client connecté sans les intercepter.
// packages/fetch-sse/server/modules/proxy.ts
import http from 'http';
import url from 'url';
const proxy = (req: http.IncomingMessage, res: http.ServerResponse) => {
const targetUrl = new URL("http://127.0.0.1:8800/stream");
const options: http.RequestOptions = {
hostname: targetUrl.hostname,
port: targetUrl.port,
path: targetUrl.pathname,
method: req.method,
headers: req.headers,
};
const proxyReq = http.request(options, (proxyRes) => {
res.writeHead(proxyRes.statusCode || 404, proxyRes.headers);
proxyRes.pipe(res); // Relai direct des données
});
// Gestion du corps de la requête POST
req.pipe(proxyReq);
proxyReq.on("error", (error) => {
console.error("Proxy error:", error);
res.writeHead(502, { "Content-Type": "text/plain" });
res.end("Bad Gateway");
});
req.socket.on("close", () => {
console.log("[proxy] connection close");
res.end();
proxyReq.destroy(); // Assurer la fermeture de la requête cible
});
};
Une attention particulière doit être portée à la gestion des événements de fermeture de connexion. L'événement socket.on('close') est généralement le plus fiable pour détecter la déconnexion du client.
Implémentation Client : Fetch API pour SSE
Utilisation de fetch pour SSE
L'utilisation de fetch pour SSE offre plusieurs avantages par rapport à EventSource : prise en charge des méthodes autres que GET, définition d'en-têtes personnalisés et gestion de la logique de reconnexion. Cela permet également d'accéder directement à l'objet Response et à son ReadableStream.
// packages/fetch-sse/client/components/fetch.tsx
import { StreamParser } from '../utils/stream-parser'; // Assumant une implémentation similaire côté client
const onOpen = (res: Response) => { /* ... */ };
const onMessage = (message: any) => { /* ... */ };
const fetchSSE = async (url: string, options?: RequestInit) => {
const signal = new AbortController();
const fetchOptions = {
...options,
method: options?.method || "GET",
signal: signal.signal,
};
try {
const res = await fetch(url, fetchOptions);
onOpen(res);
const body = res.body;
if (!body) return;
const reader = body.getReader();
const parser = new StreamParser(); // Instanciation du parser SSE
parser.onMessage = onMessage;
const process = async (): Promise<void> => {
const { done, value } = await reader.read();
if (done) return;
parser.onBinary(value); // Traitement des données binaires reçues
await process(); // Lecture récursive
};
await process();
} catch (error) {
if (error.name === 'AbortError') {
console.log('Fetch SSE aborted');
} else {
console.error("Fetch SSE error:", error);
// Logique de gestion d'erreur
}
}
// Gestion de la fermeture de la connexion pour annuler la requête
req.socket.on("close", () => { // Ceci est une simplification, devrait être géré par un hook ou un state manager
signal.abort();
});
};
</void>
Le parsing du ReadableStream reçu via fetch est similaire à l'implémentation côté serveur, utilisant une classe StreamParser pour interpréter les événements SSE.
Interaction en Flux et Rendu Dynamique
Pour une expérience utilisateur fluide, les données SSE peuvent être rendues progressivement dans le DOM. Le texte reçu doit être décodé (en remplaçant \\n par \n) et affiché avec un délai contrôlé.
// packages/fetch-sse/client/components/stream.tsx
import MarkdownIt from 'markdown-it'; // Librairie pour parser le Markdown
import { useMemoFn } from 'your-hook-library'; // Assurez-vous que cette bibliothèque est disponible
// ... (Déclarations des refs, états, currentIndex, etc.)
const append = (text: string) => {
const el = ref.current;
if (!el) return;
const mdIt = new MarkdownIt();
const textHTML = mdIt.render(text);
const dom = new DOMParser().parseFromString(textHTML, "text/html");
// Logique pour mettre à jour le DOM en ajoutant de nouvelles lignes
// et en supprimant les anciennes pour éviter de recharger tout le contenu.
const current = currentDOMIndex.current;
const children = Array.from(el.children);
for (let i = current; i < children.length; i++) {
children[i] && children[i].remove();
}
const nextNodes = dom.body.children;
for (let i = current; i < nextNodes.length; i++) {
nextNodes[i] && el.appendChild(nextNodes[i].cloneNode(true));
}
currentDOMIndex.current = nextNodes.length - 1;
};
const onMessage = useMemoFn((e: Message) => {
if (e.event !== "message") return;
setPainting(true); // Indique que l'écriture est en cours
const data = e.data.replace(/\\n/g, "\n"); // Décoder les sauts de ligne
const start = currentIndex.current;
const len = data.length;
const delay = len - start > 50 ? 10 : 50; // Vitesse d'affichage ajustée
const process = () => {
currentIndex.current++;
const end = currentIndex.current;
append(data.slice(0, end));
if (end < len) {
timer.current = setTimeout(process, delay);
} else if (!transmittingRef.current) { // Vérifier si la transmission est terminée
setPainting(false);
}
};
setTimeout(process, delay);
});
// ... (Gestion du scroll automatique et de l'effet de curseur clignotant via CSS)
La gestion du défilement automatique et l'effet de curseur clignotant peuvent être implémentés avec des écouteurs d'événements et des animations CSS pour améliorer l'expérience utilisateur.