Themen

SSE-Streams mit HTTP.sse_ticker(...)

Dieses Kapitel erklaert Server-Sent Events (SSE) in Antra von Grund auf: was SSE ueberhaupt ist, warum sse_ticker(...) anders funktioniert als ein normaler Handler, und wie du Schritt fuer Schritt einen eigenen Live-Stream baust.

Vorwissen: 13-module-und-builtins.md (Pakete und import), 07-pattern-matching.md (Tagged Values wie %Ok[...]) und 08-funktionen-lambdas-pipes.md (Lambdas). Mehr brauchst du nicht.

1. Was ist SSE ueberhaupt?#

Normales HTTP ist eine Frage und eine Antwort: Browser fragt, Server antwortet, Verbindung zu. Willst du dem Browser spaeter etwas mitteilen — ein neuer Zug auf dem Schachbrett, ein Build ist fertig, ein Fortschrittsbalken waechst — hast du drei Optionen:

1. Polling: der Browser fragt alle 2 Sekunden neu. Einfach, aber laut und traege.
2. WebSocket: volle Zwei-Wege-Verbindung. Maechtig, aber ein eigenes Protokoll.
3. SSE: der Server laesst die Antwort offen und schickt weiter Textzeilen hinterher. Einweg (Server → Browser), reines HTTP, im Browser eine Zeile Code.

SSE ist der Sweet Spot fuer "der Server hat Neuigkeiten". Antra unterstuetzt es direkt.

Wie ein SSE-Stream auf der Leitung aussieht#

Der Server antwortet einmal mit einem Header:

HTTP/1.1 200 OK
Content-Type: text/event-stream
Cache-Control: no-cache

und laesst die Verbindung dann offen. Danach schickt er, wann immer er will, kleine Bloecke:

event: board
id: 7
retry: 900
data: 7|e2e4

Jeder Block endet mit einer Leerzeile — daran erkennt der Browser, dass ein Ereignis vollstaendig ist. Zeilen, die mit : beginnen, sind Kommentare und werden ignoriert (die nutzt man als "Keepalive", damit Proxys die Leitung nicht fuer tot halten).

Im Browser:

const es = new EventSource("/stream?id=42")
es.addEventListener("board", (e) => {
    console.log("neue Daten:", e.data)   // "7|e2e4"
})

Das Schoene: EventSource reconnected von selbst, wenn die Leitung abreisst — und schickt dabei die zuletzt gesehene id: im Header Last-Event-ID mit. Darauf kommen wir in Abschnitt 6 zurueck.

2. Das Problem: ein Stream blockiert einen Server#

Die naheliegende Umsetzung waere: Handler bekommt einen Writer und schreibt in einer Schleife. Genau das ist HTTP.sse(...):

HTTP.sse(200, %{}, |writer| -> {
    while (true) {
        HTTP.sse_event(writer, latest_value())
        Time.sleep(Duration.new(:ms, 250))
    }
})

Das funktioniert — aber dieser Handler kehrt nie zurueck. Solange er laeuft, kann derselbe Prozess keine andere Verbindung bedienen. Bei einem Stream ist das egal, bei drei offenen Browser-Tabs haengt dein Server.

Man koennte pro Verbindung forken (concurrency: :fork_per_client), aber das ist ein ganzer Betriebssystem-Prozess pro Tab.

3. Die Loesung: du lieferst eine Runde, nicht die Schleife#

HTTP.sse_ticker(...) dreht die Verantwortung um. Du schreibst keine Schleife. Du schreibst eine Funktion, die einmal aufgerufen wird, sofort zurueckkehrt und sagt, was jetzt passieren soll. Die Schleife gehoert dem Server.

Das nennt man Inversion of Control — oder ganz unakademisch: "Ruf uns nicht an, wir rufen dich an."

HTTP.sse_ticker(startzustand, |zustand| -> ...eine Runde..., [optionen])

Wichtig zu verstehen: sse_ticker(...) startet gar nichts. Es baut nur einen Wert — %HttpSSETicker{...} — und gibt ihn als Antwort zurueck, genau wie HTTP.text(...) einen %HttpResponse{...} baut. Der Server sieht am Typ der Antwort: "aha, ein Stream", und uebernimmt ab da das Antreiben.

Die drei Antworten deiner tick-Funktion#

Deine Funktion bekommt den aktuellen Zustand und muss genau eines dieser Tagged Values zurueckgeben:

Gibst du etwas anderes zurueck, bricht der Stream mit einer klaren Fehlermeldung ab — das ist Absicht, ein Tippfehler soll nicht still zu einem toten Stream werden.

Die goldene Regel#

In der tick-Funktion niemals blockieren. Kein while, kein Time.sleep, kein langsamer Netzwerk-Aufruf. Sie soll in Mikrosekunden zurueckkehren.

Warum: im Multiplex-Modus teilen sich alle offenen Streams eine Schleife. Wer darin schlaeft, legt alle anderen mit lahm.

4. Dein erster Stream (vollstaendig lauffaehig)#

Ein Server, der von 1 aufwaerts zaehlt und alle 500 ms eine Zahl schickt:

import Antra::Net::HTTP as HTTP

fn handler(request) -> match request.path {
    "/stream" -> HTTP.sse_ticker(
        0,                                   # Startzustand
        |n| -> %Emit[str(n + 1), n + 1],     # eine Runde: senden, hochzaehlen
        [tick_ms: 500]
    )
    _ -> HTTP.html(200, "<script>
        new EventSource('/stream').onmessage = (e) => console.log(e.data)
    </script>")
}

HTTP.serve("127.0.0.1", 8080, handler, [concurrency: :multiplex])

Lies die tick-Funktion laut: "Bei Zustand n: sende n+1 und merke dir n+1." Kein Timing, keine Schleife, kein Socket. Genau das ist der Punkt.

Testen kannst du es ohne Browser:

curl -N http://127.0.0.1:8080/stream

-N schaltet das Puffern ab, sonst siehst du nichts tropfen.

5. Nur senden, wenn sich wirklich etwas geaendert hat#

Ein Zaehler ist langweilig; meistens willst du eine Quelle beobachten und nur bei Aenderung senden. Genau dafuer ist der Zustand da: du merkst dir darin, was der Client zuletzt bekommen hat.

Das echte Beispiel aus chess-tools (app.ant):

fn _board_tick(board_id, last_sent) -> match BoardStore.load(board_id) {
    %Ok[board] when board.version > last_sent -> %Emit[
        _board_payload(board),
        board.version,
        [event: "board", id: str(board.version), retry: 900]
    ]
    %Ok[_board] -> %Idle[last_sent]
    %Err[_] -> %Done[]
}

Drei Zeilen, drei Faelle — und match mit Guard macht sie lesbar:

• Board hat eine hoehere Version als zuletzt gesendet → senden, und die Version wird der neue Zustand.
• Board unveraendert → %Idle, Zustand bleibt.
• Board weg (geloescht, kaputt) → %Done[], Stream sauber beenden.

Der Zustand ist hier also nicht "irgendein Zaehler", sondern die Antwort auf die Frage: was hat dieser eine Client schon gesehen? Das ist das Muster, das du in 90 % aller Faelle willst.

Eingehaengt wird das so:

HTTP.sse_ticker(
    _last_seen_version(request),
    |last_sent| -> _board_tick(board_id, last_sent),
    [tick_ms: 250, keepalive_ms: 15000, max_age_ms: 300000]
)

Das Lambda ist eine Closure: board_id kommt aus dem Request und ist ab da fest verdrahtet, ohne dass es Teil des Zustands sein muss. Nur das, was sich pro Runde aendert, gehoert in den Zustand.

6. Reconnect, der nicht weh tut#

Erinnerst du dich an id: im Ereignis? Wenn die Leitung abreisst, verbindet EventSource sich neu und schickt die letzte id im Header Last-Event-ID. Liest du den als Startzustand, macht der Stream genau da weiter, wo er aufgehoert hat — kein Duplikat, keine Luecke:

fn _last_seen_version(request) -> {
    my raw = request.headers["last-event-id"]
    if (defined? raw) {
        match Int.parse(str(raw)) {
            %Ok[version] -> if (version > 0) then version else 0
            %Err[_] -> 0
        }
    } else {
        0
    }
}

Ein frischer Client hat keinen Header → Startzustand 0 → beim ersten Tick ist board.version > 0 und er bekommt sofort den aktuellen Stand. Ein wiederkehrender Client mit Last-Event-ID: 7 bekommt erst ab Version 8 wieder etwas. Dasselbe Stueck Code deckt beide Faelle ab.

Deshalb heisst der Stream in der Doku resumable: der Zustand lebt nicht im Server-Prozess, sondern kommt vom Client mit.

7. Die Optionen#

HTTP.sse_ticker(state, tick, [
    status: 200,          # HTTP-Status der Stream-Antwort
    headers: %{},         # zusaetzliche Header
    tick_ms: 250,         # Abstand zwischen zwei Runden
    keepalive_ms: undef,  # Kommentarzeile, wenn so lange nichts gesendet wurde
    max_age_ms: undef     # Stream nach dieser Zeit schliessen
])

Was die drei Zeit-Optionen wirklich bewirken:

• tick_ms — wie oft deine Funktion gefragt wird. Kleiner = frischer, aber mehr Last (bei einer Datenbank-Abfrage pro Runde spuerst du das). 250 ms sind fuer UI-Zwecke ein guter Startwert.
• keepalive_ms — wenn so lange nichts gesendet wurde, schreibt der Server von sich aus eine Kommentarzeile. Das haelt Proxys und Load-Balancer davon ab, die "stille" Verbindung zu kappen. 15000 ist ein bewaehrter Wert.
• max_age_ms — beendet den Stream absichtlich nach einer Weile. Klingt kontraproduktiv, ist aber gesund: der Browser reconnected automatisch (nach retry: Millisekunden), haengende Verbindungen raeumen sich selbst auf, und jeder Reconnect nimmt via Last-Event-ID sauber den Faden auf.

Und in opts von %Emit[...]:

• event: — der Ereignisname. Ohne ihn landet alles bei onmessage; mit ihm kannst du im Browser addEventListener("board", ...) benutzen und mehrere Ereignisarten ueber einen Stream schicken.
• id: — die Wiederaufsetz-Marke aus Abschnitt 6.
• retry: — sagt dem Browser, wie lange er vor einem Reconnect warten soll.

8. Server-Modi: warum :multiplex#

HTTP.serve(...) kennt drei Modi:

HTTP.serve("127.0.0.1", 8080, handler, [concurrency: :multiplex])

Der grosse Gewinn von sse_ticker(...): derselbe Handler laeuft in allen drei Modi. Du entwickelst mit :sequential, schaltest fuer den Betrieb auf :multiplex um, und dein Code aendert sich nicht.

Ein HTTP.sse(...) (die blockierende Variante) wird im Multiplex-Modus dagegen aktiv abgelehnt — mit einem 500er und der Meldung:

HTTP.serve: concurrency :multiplex requires HTTP.sse_ticker(...), not HTTP.sse(...)

Das ist bewusst so. Ein blockierender Callback im Reaktor wuerde alle anderen Verbindungen einfrieren, und von aussen sieht das exakt aus wie ein haengender Server — ein Fehler, den man stundenlang sucht. Lieber ein lauter Fehler als ein stiller Haenger.

9. Was der Server fuer dich uebernimmt#

Damit klar ist, was du nicht schreiben musst:

• Timing — wann deine Funktion faellig ist, und das Warten dazwischen.
• Der Response-Head — Content-Type: text/event-stream und Freunde.
• Frame-Formatierung — aus %Emit[data, ..., opts] werden die event:/id:/retry:/data:-Zeilen samt Leerzeile.
• Keepalives — die Kommentarzeilen nach keepalive_ms.
• Disconnects — geht der Browser weg, wird das erkannt und der Socket geschlossen. "Broken pipe" gilt als normales Stream-Ende, nicht als Fehler.
• Alterslimit — max_age_ms und das Aufraeumen danach.

Dein Anteil ist genau eine Frage: "Was soll in dieser Runde passieren?"

10. Typische Fehler#

Eine Schleife in der tick-Funktion. Der haeufigste. Wenn du while oder Time.sleep tippst, willst du stattdessen den Zustand nutzen: was du "beim naechsten Durchlauf" tun wolltest, gehoert in den naechsten Tick.

Immer %Emit, nie %Idle. Dann sendest du bei jedem Tick, auch wenn sich nichts geaendert hat — ein Polling-Feed mit extra Schritten. Vergleiche gegen den Zustand und antworte im Normalfall mit %Idle.

Zustand nicht weiterreichen. %Emit[data, last_sent] statt %Emit[data, board.version] sendet dieselbe Nachricht endlos, weil sich der Zustand nie bewegt. Der zweite Wert von %Emit ist immer der neue Zustand.

id: vergessen. Ohne id: schickt der Browser beim Reconnect keinen Last-Event-ID-Header, und dein Client faengt jedes Mal bei null an.

Zu kleines tick_ms bei teurer Quelle. tick_ms: 10 mit einer Datenbank-Abfrage pro Runde sind 100 Abfragen pro Sekunde pro offenem Tab.

Kurzreferenz#

# Antwort bauen (startet nichts, gibt %HttpSSETicker{...} zurueck)
HTTP.sse_ticker(state, tick)
HTTP.sse_ticker(state, tick, [status:, headers:, tick_ms:, keepalive_ms:, max_age_ms:])

# Rueckgabewerte von tick(state)
%Emit[data, next_state]
%Emit[data, next_state, [event:, id:, retry:]]
%Idle[next_state]
%Done[]

# Server starten
HTTP.serve(host, port, handler, [concurrency: :multiplex])

Wie das intern funktioniert — der Reaktor, IO.select, das Fuehren mehrerer Streams in einem Prozess — steht in dev/http-multiplex-und-ticker-interna.md.

Quelle dieser Seite