RxJS: Push und Pull endlich auseinanderhalten

Diesen Beitrag anhören (8 Min.)

Wer ein kaltes Observable wie ein geteiltes Promise behandelt, schickt seine HTTP-Anfrage so oft los, wie jemand subscribe aufruft. Drei Komponenten am selben kalten Stream bedeuten dann drei Requests – ohne Fehlermeldung, ohne Warnung, exakt nach Definition.

Der Grund dafür hat zunächst weniger mit Operatoren zu tun als mit einer Frage: Wer entscheidet, wann ein Wert fließt – der Producer oder der Consumer? Vor map, switchMap und debounceTime wird meist gefragt, welcher Operator der richtige ist. Diese Frage kommt zu früh. Push und Pull liefern das Grundmodell, zu dem später Cold, Hot und Multicasting kommen.

Die eine Frage, die alles ordnet

Jede Datenquelle in JavaScript lässt sich über ihre Richtung beschreiben. Es gibt nur zwei.

Bei einem Pull-System holt sich der Consumer die Werte aktiv. Er ruft etwas auf und bekommt eine Antwort. Eine Funktion ist genau das: Du rufst sie, sie liefert. Ein Iterator ist dasselbe in mehreren Schritten, denn iterator.next() zieht den nächsten Wert heraus. Der Consumer bestimmt das Tempo.

Bei einem Push-System dreht sich das um. Jetzt entscheidet der Producer, wann etwas geschickt wird, und der Consumer weiß vorher nicht, wann der nächste Wert kommt. Ein Klick-Event, eine WebSocket-Nachricht, ein Timer: All das schiebt, sobald es so weit ist. Der Consumer reagiert nur noch.

Das klingt akademisch, hat aber direkte Folgen für den Code. Bei synchronem Pull bestimmt der Consumer den nächsten Abruf und kann dabei blockieren; ein asynchroner Iterator muss das nicht tun. Bei Push entscheidet der Producer, wann er liefert, und der Consumer reagiert darauf. Genau hier scheitern viele Migrationen von „Daten abfragen“ zu „auf Daten reagieren“.

Vier Produzenten, zwei Richtungen

Die offizielle RxJS-Dokumentation ordnet die vier wichtigen Produzenten in ein Raster ein. Ich male es in Architektur-Diskussionen als Erstes ans Whiteboard, weil danach die meisten Fragen schon beantwortet sind.

Producer	Anzahl Werte	Richtung	Auswertung
Function	genau ein Wert	Pull	lazy
Iterator / Generator	null bis viele	Pull	lazy
Promise	genau ein Wert	Push	eager
Observable	null bis viele	Push	lazy

Lies die Tabelle als zwei Achsen. Die eine Achse fragt: ein Wert oder mehrere? Die andere fragt: zieht der Consumer oder schiebt der Producer? Das Observable ist die einzige Quelle, die mehrere Werte schiebt und dabei trotzdem lazy bleibt. Diese Kombination macht es so mächtig und für Einsteiger so verwirrend, weil sie keine direkte Entsprechung im klassischen JavaScript hat.

Ein Promise ist der nächste Verwandte, deckt aber nur eine Zelle daneben ab: ein Wert, geschoben, sofort gestartet. Wer das verinnerlicht, hört auf, ein Observable wie ein Promise mit mehr Methoden zu behandeln.

Ein Observable ist eine Funktion mit mehreren Ausgängen

Am schnellsten wird das im Code klar. Mein introduction-rxjs-Material beginnt bewusst beim Observable-Konstruktor, bevor es Operatoren zeigt. Er zwingt dich, Producer und Consumer selbst zu benennen.

import { Observable } from 'rxjs';

// The producer decides on its own when values flow.
const ticks$ = new Observable((observer) => {
  let n = 0;
  const id = setInterval(() => observer.next(n++), 1000);

  // Cleanup function: runs on unsubscribe or on complete.
  return () => clearInterval(id);
});

// The consumer only reacts, it does not actively pull.
const sub = ticks$.subscribe({
  next: (v) => console.log('value', v),
  error: (e) => console.error('error', e),
  complete: () => console.log('done'),
});

Der Block in subscribe ist der Observer mit seinen drei Kanälen: next für Werte, error für den Abbruch mit Fehler, complete für das saubere Ende. Das ist das vollständige Vokabular eines Push-Consumers. Mehr gibt es nicht, und mehr braucht man auch nicht.

Zum Vergleich dieselbe Idee als Pull-Quelle. Hier passiert nichts, solange niemand zieht.

function* naturals() {
  let n = 0;
  while (true) yield n++; // The value is produced only when pulled.
}

const it = naturals();
it.next().value; // 0  – the consumer pulls
it.next().value; // 1  – and sets the pace

Beide Quellen liefern eine unendliche Folge natürlicher Zahlen. Der Unterschied liegt allein in der Kontrolle. Beim Generator entscheidest du mit jedem next(), wann die Eins nach der Null kommt. Beim Observable entscheidet der Timer, und du bekommst die Werte einfach zugestellt.

Die folgenreichste Spalte der Tabelle ist „Auswertung“. Das hier konstruierte kalte Observable ist lazy. Es führt seinen Producer-Code erst aus, wenn subscribe aufgerufen wird, und für jedes subscribe von vorn. Heiße Observables und geteilte Quellen verhalten sich anders. Ein Promise ist dagegen eager: Es startet seine Arbeit in dem Moment, in dem es erzeugt wird, ganz ohne then.

An diesem Unterschied entstehen die meisten Fehler beim Umstieg.

// eager: the request goes out immediately, even without then.
const preise = fetch('/api/preise');

// lazy: nothing happens until someone calls subscribe.
const preise$ = new Observable((observer) => {
  fetch('/api/preise')
    .then((res) => res.json())
    .then((data) => {
      observer.next(data);
      observer.complete();
    })
    .catch((err) => observer.error(err));
});

Damit lässt sich der Fall vom Anfang auflösen. Ein Team baut eine Suche um. Vorher gab es ein geteiltes Promise, das einmal lief und dessen Ergebnis sich mehrere Komponenten teilten. Nach der Umstellung auf ein Observable abonnieren drei Komponenten denselben Stream. Plötzlich gehen drei HTTP-Anfragen raus statt einer. Niemand hat einen Fehler gemacht im klassischen Sinn, das Observable verhält sich exakt nach Definition: lazy und kalt, also pro Subscription eine eigene Ausführung. Wer das Promise-Verhalten 2017 zurückhaben wollte, teilte den Stream mit publishReplay und refCount oder setzte ein Subject davor. publishReplay ist inzwischen veraltet; heute drückt shareReplay({ bufferSize: 1, refCount: true }) dieselbe Absicht direkter aus. Bewusst entscheiden muss man das Multicasting weiterhin. Erst wenn man Push, Lazy und Cold getrennt benennt, sieht man, warum hier drei Requests entstehen.

Zeit sichtbar machen mit dem Marble-Diagramm

Bei Push-Systemen hängt alles am zeitlichen Verlauf, und genau das macht sie schwer zu lesen. Ein Marble-Diagramm zeichnet diese Zeit auf eine Linie. Jede Murmel ist ein gepushter Wert, das | ist complete, das X wäre ein error.

subscribe                         unsubscribe
   |                                   |
   v                                   v
ticks$: ---0----1----2----3----4----5--|
              (the producer pushes every 1000 ms)

Die Linie macht zwei Dinge sofort sichtbar, die im Text untergehen. Erstens: Die Werte erscheinen nach der inneren Logik des Producers, hier dem Intervall, und eben nicht auf Abruf. Zweitens: Es gibt zwei Ränder, die Subscription und die Unsubscription. Zwischen diesen Rändern existiert der Stream, davor und danach nicht. Wer Marble-Diagramme liest, hört auf, über einzelne Werte nachzudenken, und beginnt über die Lebensdauer des Streams nachzudenken.

Für die Richtung selbst genügt ein kleines Schema. Es zeigt, wer wen antreibt.

flowchart LR
  subgraph Pull
    C1[Consumer] -- ruft next ab --> P1[Producer]
  end
  subgraph Push
    P2[Producer] -- schiebt next --> C2[Consumer]
  end

Der Fehler, den fast jeder einmal baut

Aus der Lazy-Eigenschaft folgt eine Pflicht, die im Material ein eigenes Kapitel hat, nämlich Disposable. Wenn ein subscribe einen Producer startet, dann muss irgendwer ihn auch wieder beenden. Beim ticks$ oben ist das offensichtlich: Der setInterval läuft ewig weiter, solange niemand unsubscribe aufruft.

const sub = ticks$.subscribe((v) => console.log(v));

// Without this line the timer ticks forever,
// even after the component is long gone.
sub.unsubscribe();

In einer Single-Page-Anwendung kann daraus ein echtes Leck werden. Eine Komponente abonniert einen Event-Stream, der Nutzer navigiert weg, die Komponente verschwindet aus dem DOM, aber ein nicht beendetes Abo hält Listener und referenzierte Objekte am Leben. Beim nächsten Öffnen kommt ein zweites Abo dazu, dann ein drittes. Die Symptome sind doppelt ausgeführte Nebenwirkungen und wachsender Speicherverbrauch. Die Subscription ist die Grenze, an der ein Stream Besitz und Lebensdauer bekommt; je nach Quelle endet sie durch complete, error oder ein bewusstes Unsubscribe.

Was bleibt

Push und Pull ist kein theoretisches Detail, sondern das Raster, an dem du jede Entscheidung in RxJS aufhängst. Brauche ich einen oder mehrere Werte? Zieht mein Consumer oder schiebt mein Producer? Startet die Arbeit sofort oder erst beim Abonnieren? Und wer beendet den Stream wieder? Diese vier Fragen klären die meisten Operator-Diskussionen, bevor sie entstehen.

Die Reihenfolge in meinem Begleitmaterial – das ich auch mit Partnern wie Grossweber einsetze – folgt genau dieser Logik. Erst Observable, Observer und Notification, also das Push-Vokabular. Dann Scheduler und Disposable, also Zeit und Lebensdauer. Operatoren kommen zuletzt, weil sie nur Werkzeuge auf diesem Fundament sind. Wer mit der Operatorliste anfängt, lernt Vokabeln ohne Grammatik.

Weiterführende Quellen

RxJS-Dokumentation, Kapitel Observable mit der Push/Pull-Tabelle: https://rxjs.dev/guide/observable
RxJS: publishReplay und heutige Alternativen
RxJS: shareReplay
Begleitmaterial der Schulung, Repository introduction-rxjs: https://github.com/MikeBild/introduction-rxjs

Wie fandest du diesen Beitrag?

Kommentare