← Zurück zum Blog

Tech

Streaming mit Server-Sent Events (SSE)

5 Min. Lesezeit

Wenn man Chat-Funktionalitäten mit KI programmiert, sieht es immer toll aus, wenn man das Streaming der Modelle auch ausnutzt und den Stream, so wie er von der KI kommt, direkt ausgibt. Das sieht dann aus, als ob das Gegenüber wirklich tippen würde.

Mein erster Gedanke war, dass man das sicherlich prima mit SignalR lösen kann. Dann habe ich aber ein anderes, tolles Feature gefunden, das dafür viel besser geeignet ist: SSE (Server-Sent Events).

Es handelt sich dabei um einen Web-Standard aus der HTML5-Spezifikation, der es einem Server erlaubt, über eine einzelne, offene HTTP-Verbindung kontinuierlich Daten an einen Client zu schicken. Das geht nur unidirektional, also vom Server zum Client — aber genau das ist es, was ich hier brauche. Und da es ein Web-Standard ist, wird das von jedem modernen Browser unterstützt.

Also: Wie wird das implementiert?

API-Implementierung

Fangen wir auf der Serverseite an.

Ganz grob betrachtet startet man SSE einfach mit einem speziellen Header. Mit dem Content-Type text/event-stream sagt man dem Client nämlich, dass hier kein normaler Response kommt, sondern ein Event-Stream. Erst durch diesen Header weiß der Browser, dass er die Antwort nicht als Ganzes abwarten soll, sondern dass kontinuierlich weitere Daten über die Verbindung nachkommen können.

Zusätzlich setzt man in der Regel noch weitere Header:

  • Cache-Control: no-cache — damit Browser, Proxy, CDN oder was auch immer nicht auf die Idee kommen, die Antwort zwischenzuspeichern. Das ist wichtig, denn die Antwort endet ja eigentlich nie, sondern ist ein stetig weiterlaufender Stream.
  • Connection: keep-alive — damit wird sichergestellt, dass die Verbindung nicht einfach geschlossen wird. Für SSE muss die Verbindung dauerhaft offen gehalten werden, da sonst der Stream abbricht.
  • X-Accel-Buffering: no — das ist eigentlich nichts Offizielles. Dieser Header ist nginx-spezifisch und wird auch nur benötigt, wenn die API hinter einem nginx-Proxy läuft. Da ich eigentlich immer mit nginx arbeite, habe ich ihn hier gleich mit aufgenommen. Hintergrund ist, dass nginx Responses standardmäßig puffert. Das heißt, die Zeichen würden nicht einzeln am Client ankommen, sondern alle auf einmal — dann, wenn nginx es beschließt.

Wie geht es dann in der API weiter? Eigentlich nichts Dramatisches: Man schreibt das, was man in den Stream schicken will, einfach in den Response-Body. Nach jedem Schreiben in den Body sollte man gleich ein Flush aufrufen, damit die Daten auch sicher sofort rausgehen. Und das war's eigentlich schon.

Ach ja: Manchmal kann es sinnvoll sein, den Stream am Ende sauber zu schließen. Das macht man, indem man event: done in den Stream schickt. Der Client erkennt das und schließt die Verbindung dann ebenfalls. Das ist aber optional — man kann den Stream auch einfach offen lassen.

Hier der Beispielcode dafür:

[HttpGet("stream-text")]
public async Task StreamText(CancellationToken cancellationToken)
{
    Response.ContentType = "text/event-stream";
    Response.Headers.Append("Cache-Control", "no-cache");
    Response.Headers.Append("Connection", "keep-alive");
    Response.Headers.Append("X-Accel-Buffering", "no");

    try
    {
        foreach (var character in _text)
        {
            if (cancellationToken.IsCancellationRequested)
                return;

            var payload = JsonSerializer.Serialize(new { character = character.ToString() });
            await Response.WriteAsync($"data: {payload}\n\n", cancellationToken);
            await Response.Body.FlushAsync(cancellationToken);
            await Task.Delay(100, cancellationToken);
        }

        // Optional: Abschluss-Event, damit der Client weiß: "fertig"
        await Response.WriteAsync("event: done\ndata: {}\n\n", cancellationToken);
        await Response.Body.FlushAsync(cancellationToken);
    }
    catch (OperationCanceledException)
    {
        // Client hat die Verbindung getrennt => kein Fehler
    }
}

Ein kleiner Hinweis zum Format: Jede Nachricht ist eine Zeile, die mit data: beginnt und mit einer Leerzeile (\n\n) abgeschlossen wird. Genau diese Leerzeile signalisiert das Ende eines Events. Über eine vorangestellte event:-Zeile lässt sich ein Event zusätzlich benennen — das nutzen wir oben für das done-Event.

Client-Implementierung

In Angular verwendet man — wie sollte es auch anders sein — Observables. Aber lösen wir die Implementierung auch hier komplett auf.

Ich verwende dafür eigentlich immer einen kleinen Hilfs-Service, der sich um SSE kümmert. Die Methode ist ganz unspektakulär: Sie erstellt ein Observable, im Observable wird dann ein EventSource-Objekt angelegt — und das ist im Großen und Ganzen schon die ganze Magie.

Hier der Code für diesen Hilfs-Service:

@Injectable({ providedIn: 'root' })
export class SseService {
    /**
     * Liefert jedes `data:`-Event der SSE-Verbindung als geparstes JSON,
     * bis der Server `event: done` schickt.
     */
    stream<T>(url: string): Observable<T> {
        return new Observable<T>((subscriber) => {
            const eventSource = new EventSource(url);

            eventSource.onmessage = (event) => subscriber.next(JSON.parse(event.data) as T);

            eventSource.addEventListener('done', () => subscriber.complete());

            eventSource.onerror = () => {
                if (eventSource.readyState === EventSource.CLOSED) {
                    subscriber.error(new Error(`SSE-Verbindung zu ${url} fehlgeschlagen`));
                }
            };

            return () => eventSource.close();   // läuft bei complete, error UND unsubscribe
        });
    }
}

Ein paar Worte dazu: onmessage wird nur für die namenlosen (Standard-)Events ausgelöst — also für unsere data:-Zeilen mit dem eigentlichen Text. Das benannte done-Event fängt deshalb ein eigener addEventListener('done', …) ab und schließt den Stream über subscriber.complete(). Die Aufräum-Funktion, die das Observable am Ende zurückgibt, ruft eventSource.close() auf — und zwar zuverlässig bei complete, bei error und wenn sich der Aufrufer vorzeitig abmeldet (unsubscribe).

In der Methode, in der man diesen Helper dann aufruft, holt man sich die Subscription, konfiguriert sie mit dem entsprechenden Endpunkt und reagiert auf die Events, die daraus kommen. Hier ein Beispiel:

this.subscription = this.sseService
    .stream<CharacterEvent>('/api/sse/stream-text')
    .pipe(map((event) => event.character))
    .subscribe({
        next: (char) => this.displayedText.update((text) => text + char),
        error: (err) => {
            console.error('SSE-Fehler:', err);
            this.isStreaming.set(false);
        },
        complete: () => {
            this.isStreaming.set(false);
        }
    });

Das heißt: Im next bekomme ich jedes Event und hänge es einfach an den bereits bestehenden Text an. Ein bisschen Fehlerbehandlung noch dazu — und das war dann auch schon alles.

Fazit

Ich war ehrlich überrascht, wie wenig man für dieses Feature eigentlich braucht. Kein extra Nuget-Paket, keine zusätzliche Bibliothek im Client, keine WebSocket- oder SignalR-Infrastruktur — ein Header auf der Serverseite und ein EventSource im Browser, und schon läuft der Stream. Für genau den Fall, den ich hatte — Daten in eine Richtung vom Server zum Client schieben —, ist SSE deutlich schlanker als die schwereren Alternativen.

Man sollte sich nur bewusst sein, wo die Grenzen liegen: EventSource kann ausschließlich GET, und sobald man echte Zwei-Wege-Kommunikation braucht, ist man mit WebSockets bzw. SignalR besser bedient. Für den „KI tippt gerade"-Effekt aber ist SSE für mich seitdem das Mittel der Wahl.