Web Messaging mit RxJS

var incoming = Rx.Observable.create(function (observer) {
  var source = new EventSource('/subscribe/green');

  source.onmessage = function (event) {
    observer.onNext(JSON.parse(event.data));
  };
  source.onerror = function (error) {
    observer.onError(error);
  };

  return function () {
    source.close();
  };
});

incoming
  .map(function (message) { return message.green; })
  .distinctUntilChanged()
  .subscribe(function (count) {
    render(count);
  });

Das Interessante an diesem Stück Client-Code ist der Teil, der fehlt. Die Konsumseite – die drei Zeilen ab incoming.map(...) – weiß nichts über EventSource, nichts über HTTP, nichts über Verbindungen. Sie abonniert einen Strom eingehender Nachrichten und verarbeitet ihn mit denselben Operatoren, mit denen sie auch ein Array oder eine Folge von Mausklicks verarbeiten würde. Wie die Nachrichten technisch ins Haus kommen, ist im create-Callback eingesperrt. Genau diese Trennung ist der Kern dessen, was ich in den letzten Wochen ausprobiert habe: Nachrichten zwischen Browser und Server nicht als Verbindungsproblem denken, sondern als Strom – mit RxJS als gemeinsamer Sprache auf beiden Seiten.

Warum nicht gleich WebSockets?

Sobald in einem Projekt das Wort „Echtzeit" fällt, liegt der Griff zu WebSockets nahe – meist in Form von socket.io, das die Verbindungsdetails freundlich wegkapselt. Das ist oft die richtige Wahl, aber sie hat einen Preis, der gern erst später auffällt. WebSockets sind ein eigenes Protokoll mit eigenem Handshake. Ältere Proxies verschlucken das Upgrade, manche Firmen-Firewalls kappen langlebige Verbindungen kommentarlos, Load Balancer brauchen Sonderbehandlung, und die gewohnte HTTP-Werkzeugkiste – Caching, Auth-Header, Access-Logs, curl zum Debuggen – greift nur noch teilweise. socket.io beantwortet einen Teil davon mit Fallbacks, holt sich dafür aber eine ganze Bibliotheksschicht auf Client und Server ins Projekt.

Dabei ist die fachliche Anforderung häufig bescheidener, als das Wort „Echtzeit" klingt: Der Server soll den Browser über Ereignisse informieren, überwiegend in eine Richtung, und eine Latenz von Sekundenbruchteilen genügt völlig. Für dieses Profil hat HTTP seit Jahren Antworten, die ohne Protokollwechsel auskommen: Long-Polling, chunked Responses, Server-Sent Events. Alles davon läuft durch jede Infrastruktur, die HTTP kann – also durch praktisch alles.

Die eigentliche Architekturfrage ist deshalb nicht „WebSockets oder nicht?", sondern: Wie schreibe ich den Anwendungscode so, dass diese Entscheidung später revidierbar bleibt?

Ein Experiment: alles ist ein Strom

Vor knapp zwei Wochen habe ich dazu ein kleines Repository veröffentlicht, web-messaging. Die README-Zeile beschreibt den Umfang ehrlich: „Message routing and Complex-Event-Streams via HTTP (Transfer-Encoding: chunked)". Es ist ein Experiment, kein Produkt – eine einzelne server.js, eine index.html, kein Express, nicht einmal ein Routing-Framework, sondern ein handgeschriebener Mini-Router aus regulären Ausdrücken. Gerade diese Schlichtheit macht sichtbar, worum es geht.

Der Server modelliert alles, was Nachrichten erzeugt, als Observable. Eine simulierte Quelle tickt alle zwei Sekunden und würfelt eine Gruppe; ein Subject dient als Eingang für Nachrichten von außen. Beides fließt in einen gemeinsamen Strom:

var rx = require('rx');

var publishChannel = new rx.Subject();

var simulated = rx.Observable.interval(2000)
  .map(function (tick) {
    return { sequence: tick, group: randomGroup() };
  });

var events = simulated
  .merge(publishChannel)
  .publish()
  .refCount();

var totals = events
  .scan({}, function (acc, message) {
    acc[message.group] = (acc[message.group] || 0) + 1;
    return acc;
  })
  .publish()
  .refCount();

Zwei Dinge daran haben sich als tragend erwiesen. Erstens merge: Dass eine simulierte Zeitquelle und ein von HTTP-POSTs gefüttertes Subject sich zu einem Strom vereinigen lassen, ohne dass irgendwer davon erfährt, ist keine Spielerei – es heißt, dass „woher kommt die Nachricht?" eine lokale Entscheidung bleibt. Zweitens publish().refCount(): Ohne diese beiden Aufrufe wäre der Strom kalt, und jeder Abonnent bekäme seine eigene Simulation mit eigenem Zähler. So aber teilen sich alle Verbindungen dieselbe heiße Quelle, und scan führt genau einen Zählerstand über alle Gruppen – Zustand als Strom, nicht als Variable, an der viele Stellen herumschreiben.

Der publishChannel wird von der POST-Route gefüttert, und das ist die vielleicht hübscheste Zeile des Experiments: Ein gewöhnlicher HTTP-Request wird zu einem Ereignis im Strom.

route.post(/\/publish\/([a-z]*)/, function (req, res) {
  publishChannel.onNext({ group: req.params[0] });
  res.end();
});

Die Antwort als Observer

Spannender als die Erzeugerseite ist die Auslieferung. Wie kommt ein Server-Strom in einen Browser, ohne WebSockets? Die Antwort im Experiment: Die HTTP-Response bleibt einfach offen. Mit Transfer-Encoding: chunked darf ein Server beliebig lange häppchenweise schreiben – und damit wird res zu einem Observer, der Nachrichten aus dem Strom entgegennimmt:

route.get(/\/subscribe\/([a-z]*)/, function (req, res) {
  res.setHeader('Content-Type', 'application/json');
  res.setHeader('Transfer-Encoding', 'chunked');

  var subscription = totals
    .map(function (data) {
      var group = req.params[0];
      if (!group) return data;
      var result = {};
      result[group] = data[group];
      return result;
    })
    .subscribe(function (data) {
      res.write(JSON.stringify(data) + '\n\n');
    });

  req.on('close', function () {
    subscription.dispose();
  });
});

Jede Verbindung ist eine Subscription. Das klingt banal, erledigt aber nebenbei das Problem, das in handgestrickten Publish-Subscribe-Servern regelmäßig zu Speicherlecks führt: die Lebensdauer. Wenn der Browser die Verbindung schließt – Tab zu, Netz weg, Seite neu geladen – feuert req.on('close'), die Subscription wird mit dispose() beendet, und weil refCount mitzählt, stellt die Quelle die Arbeit ein, sobald der letzte Abonnent weg ist. Aufräumen ist keine Sonderlogik mehr, sondern die natürliche Kehrseite des Abonnierens. In einem Projekt einige Jahre zuvor haben wir für genau dieses Aufräumen eine eigene Registry mit Timeout-Heuristiken gepflegt – hier sind es zwei Zeilen.

Ein Nebeneffekt dieses Aufbaus, den ich nicht mehr hergeben möchte: Der Strom lässt sich mit Bordmitteln beobachten. Ein curl http://localhost:8765/subscribe/green im Terminal, und die Nachrichten tropfen zeilenweise herein – kein Spezialwerkzeug, kein Browser-Plugin, keine Protokoll-Inspektion. Das Repository hat dafür sogar eine eigene Route /log, die den ungefilterten Rohstrom vor der Aggregation ausliefert. Beim Entwickeln hatte ich ständig zwei Terminals offen: eines auf /log für die Rohereignisse, eines auf /subscribe/... für das, was ein Client tatsächlich sieht. Bei einem WebSocket-Aufbau braucht dieselbe Sichtbarkeit bereits eigenen Werkzeugbau oder zumindest die Frame-Ansicht der Browser-Entwicklerwerkzeuge.

Der Fluss im Überblick:

flowchart LR
  POST[HTTP POST<br/>/publish/:group] --> SUBJ[Subject<br/>publishChannel]
  SIM[Interval<br/>Simulation] --> MERGE[merge]
  SUBJ --> MERGE
  MERGE --> SCAN[scan<br/>Zählerstand je Gruppe]
  SCAN --> SUBA[Subscription<br/>Verbindung A]
  SCAN --> SUBB[Subscription<br/>Verbindung B]
  SUBA --> RESA[chunked Response A<br/>res.write]
  SUBB --> RESB[chunked Response B<br/>res.write]

Auf der Client-Seite bin ich im Repository übrigens weniger konsequent: Die index.html ist eine kleine AngularJS-1.4-Anwendung, die die Chunks noch von Hand aus einem XMLHttpRequest klaubt – bei readyState === 3 den responseText an den doppelten Zeilenumbrüchen zerschneiden und die letzte vollständige Zeile parsen. Das funktioniert, ist aber genau die Sorte Transportdetail, die im Anwendungscode nichts verloren hat. Der Rx.Observable.create-Wrapper vom Anfang ist die Konsequenz daraus: einmal geschrieben, versteckt er dieses Gefummel, und alles dahinter spricht nur noch Observable. Wer statt roher Chunks Server-Sent Events verwendet – dasselbe Muster, nur als text/event-stream formalisiert – bekommt mit EventSource sogar Parsing und Wiederverbinden vom Browser geschenkt.

Reconnect und Backpressure als Streams-Probleme

Zwei Themen entscheiden darüber, ob so ein Aufbau ein Wochenend-Experiment bleibt oder Betrieb aushält. Beide lassen sich als Stream-Probleme formulieren, und das ist der zweite Grund, warum sich RxJS hier lohnt.

Erstens der Wiederaufbau der Verbindung. Netzverbindungen reißen – die Frage ist nicht ob, sondern wie oft. Mit EventSource verbindet der Browser von sich aus neu und schickt per Last-Event-ID sogar mit, wo er stehengeblieben ist, sofern der Server IDs vergibt. Beim rohen XHR-Ansatz baut man das selbst – und statt einer Schleife mit Zustandsvariablen wird daraus ein Operator auf dem Strom:

var resilient = incoming.retryWhen(function (errors) {
  return errors.delay(3000);
});

Der Fehlerfall ist damit kein Sonderpfad mit eigenem Code, sondern Teil derselben Deklaration wie der Erfolgsfall. Wer klügeres Verhalten will – wachsende Wartezeiten, Aufgeben nach n Versuchen – komponiert es aus denselben Bausteinen, statt eine Zustandsmaschine zu pflegen.

Zweitens Backpressure. Der Server im Experiment schreibt so schnell in die Response, wie der Strom liefert. Was, wenn der Abnehmer nicht hinterherkommt – ein Mobilgerät im Funkloch, ein überlasteter Tab? res.write gibt in Node ehrlich false zurück, wenn der Kernel-Puffer voll ist; das Experiment ignoriert das großzügig, und genau da hört der Spaß auf. Die saubere Antwort ist wieder eine Stream-Antwort, denn RxJS bringt die Werkzeuge mit, die Rate vor dem Draht zu formen:

• sample schickt in festen Abständen nur den jüngsten Stand – für Aggregationen wie den Zählerstand hier ideal, denn veraltete Zwischenstände interessieren niemanden.
• bufferWithTime sammelt Nachrichten und verschickt sie paketweise – wenn nichts verloren gehen darf, aber die Frequenz sinken soll.
• pausableBuffered hält den Strom an, bis der Abnehmer wieder Luft hat – echtes Anhalten statt Verwerfen, etwa gekoppelt an das drain-Ereignis der Response.

Wichtig ist weniger, welcher der drei Wege gewählt wird, als dass die Frage überhaupt eine Antwort im Modell hat. In der Callback-Welt ist „Erzeuger schneller als Verbraucher" ein Querschnittsproblem, das sich durch den ganzen Code zieht. Im Strom ist es eine Zeile an der richtigen Stelle.

Der Transport wird austauschbar

Damit zum eigentlichen Architekturgewinn. Wenn die Konsumseite nur ein Observable sieht, ist der Transport eine Implementierung dieses Interfaces – und Implementierungen kann man tauschen. Dieselbe Anwendung, morgen über WebSockets:

var incoming = Rx.Observable.create(function (observer) {
  var socket = new WebSocket('ws://localhost:8765/subscribe/green');

  socket.onmessage = function (event) {
    observer.onNext(JSON.parse(event.data));
  };
  socket.onerror = function (error) {
    observer.onError(error);
  };

  return function () {
    socket.close();
  };
});

Die Zeilen ab incoming.map(...) bleiben unverändert, ebenso retryWhen und alles, was darauf aufbaut. Long-Polling passt in denselben Rahmen: ein Request, der beim Eintreffen einer Nachricht antwortet, per flatMap und Wiederholung zu einem endlosen Strom verkettet. Der Wechsel des Transports ist dann ein lokaler Eingriff in einer Datei – keine Operation am offenen Herzen der Anwendung.

flowchart LR
  LP[Long-Polling] --> OBS[Observable<br/>gemeinsame Schnittstelle]
  SSE[Server-Sent Events] --> OBS
  CH[chunked HTTP] --> OBS
  WS[WebSocket] --> OBS
  OBS --> APP[Anwendungscode<br/>map / filter / scan / retryWhen]

Mir ist an diesem Punkt aufgefallen, dass die Diskussion „WebSockets vs. HTTP" in Teams oft deshalb so zäh ist, weil sie als Einbahnstraße geführt wird – als müsste die Entscheidung für immer gelten. Sobald der Transport hinter einem Observable liegt, verliert sie diese Schwere. Man darf mit dem anspruchslosesten Transport anfangen, der die Anforderung erfüllt, und später umziehen, wenn Messungen es verlangen.

Wann schlichtes HTTP reicht

Bleibt die ehrliche Abwägung. Nach diesem Experiment würde ich gewöhnliches HTTP – chunked oder als Server-Sent Events – überall dort wählen, wo drei Bedingungen zusammenkommen:

• Die Nachrichten fließen ganz überwiegend vom Server zum Browser; die Gegenrichtung sind gelegentliche Kommandos, für die ein POST völlig genügt.
• Latenzen von einigen hundert Millisekunden stören niemanden – Dashboards, Benachrichtigungen, Fortschrittsanzeigen, Feeds.
• Die vorhandene HTTP-Infrastruktur – Proxies, Auth, Logging, Debugging mit curl – soll ohne Sonderwege weiterarbeiten.

WebSockets verdienen ihren Platz, wo diese Bedingungen kippen: echter bidirektionaler Verkehr mit vielen kleinen Nachrichten in beide Richtungen, kollaboratives Editieren, Spiele, alles mit spürbarer Latenzanforderung. Dann trägt ein POST je Nachricht nicht mehr, und der Protokollwechsel zahlt sich aus. Ein Wermutstropfen der HTTP-Seite gehört ebenfalls auf den Tisch: Browser deckeln die Zahl gleichzeitiger Verbindungen pro Host, sechs sind üblich – wer viele Streams parallel öffnet, muss sie serverseitig zu einem multiplexen. HTTP/2 ist seit Mai als RFC verabschiedet und wird genau dieses Problem entschärfen, aber im Browser-Alltag ist es noch keine Größe, mit der man heute planen kann.

Und natürlich bleibt das Experiment ein Experiment: keine Authentifizierung, keine Persistenz, keine Nachrichten-IDs für lückenloses Aufholen nach Reconnect, und ein Nachrichtenraum, der genau so lange existiert wie der eine Node-Prozess. Wer mehr braucht, hängt an das Subject eine echte Quelle – eine Queue, einen Log – und genau dafür ist die Stelle ja schon vorbereitet.

Unterm Strich

Der Wert dieses kleinen Servers liegt nicht im Transport-Trick mit chunked Responses – den gab es lange vor RxJS. Er liegt darin, dass beide Seiten dieselbe Abstraktion sprechen. Auf dem Server sind Simulation, POST-Eingang und Aggregation Observables, die sich mit merge und scan komponieren lassen; im Browser ist der eingehende Nachrichtenstrom ein Observable, auf dem map, distinctUntilChanged und retryWhen arbeiten. Verbindungsabbruch, Wiederholung, Ratenbegrenzung, Aufräumen – all das, was in Callback-Code als Sonderfall über die Datei verstreut liegt, hat im Strom einen benannten Platz.

ES2015 ist seit ein paar Wochen verabschiedet, und vieles am Code oben wird mit Arrow Functions hübscher werden. Am Kern ändert das nichts: Die Entscheidung, eingehende Nachrichten als Strom zu modellieren und den Transport dahinter zu verstecken, ist unabhängig von Syntax und übrigens auch von RxJS selbst – sie würde mit jeder Observable-Implementierung funktionieren. Was ich aus dem Experiment mitnehme, ist deshalb weniger eine Bibliotheksempfehlung als eine Reihenfolge: erst das Nachrichtenmodell als Strom festlegen, dann den anspruchslosesten Transport wählen, der die Anforderung erfüllt – und die Freiheit behalten, ihn zu wechseln, wenn die Wirklichkeit es verlangt.

Weiterführende Quellen

• MikeBild/web-messaging – das besprochene Experiment: Message-Routing und Event-Streams über HTTP mit chunked Transfer-Encoding
• Reactive-Extensions/RxJS – die Reactive Extensions für JavaScript mit Dokumentation aller Operatoren
• Server-sent events (MDN) – EventSource und das text/event-stream-Format
• ReactiveX: Einführung – das Observable-Modell und die Operator-Idee sprachübergreifend erklärt

Wie fandest du diesen Beitrag?

Kommentare