Fortgeschrittenes RxJS: Operatoren höherer Ordnung

Signals haben den Zustand übernommen, aber Ereignisse über die Zeit bleiben das Revier von RxJS: Tastenanschläge, Klicks, sich überlappende Requests, Retries, Websockets. Der wichtigste Teil, und der am häufigsten falsch gemachte, sind die Higher-Order-Operatoren. Jeder Wert startet ein neues Observable, meist einen HTTP-Request. Die eigentliche Frage lautet: Was tust du, wenn der nächste Wert eintrifft, während der vorherige Request noch läuft?

Es gibt vier Antworten: abbrechen, parallel ausführen, in die Warteschlange stellen oder ignorieren. Jede hat ihren Operator. Wählst du falsch, entsteht ein Race-Condition-Bug: alte Ergebnisse überschreiben die neuen, Bestellungen werden doppelt abgeschickt, Speichervorgänge laufen in der falschen Reihenfolge.

Das Problem: Observable von Observables

Ein map, das einen Request zurückgibt, erzeugt ein Observable<Observable<T>>: einen Stream von Requests, die niemand abonniert hat. Die *Map-Operatoren abonnieren die inneren Streams selbst und flachen sie zu einem einzigen Stream von Ergebnissen ab. Der Unterschied zwischen ihnen ist nur die Überlappungsstrategie.

// „Für jeden Wert ein Observable starten“ — die Frage ist, was du tust, wenn der nächste Wert eintrifft,
// während der vorherige noch läuft.
clicks$.pipe(map((id) => http.get(`/api/items/${id}`)));       // ✗ Observable<Observable<Item>>
clicks$.pipe(switchMap((id) => http.get(`/api/items/${id}`))); // ✓ Observable<Item> — plus eine Strategie

Die vier Operatoren

// Typeahead: Nur der letzte Request zählt; die alten werden ABGEBROCHEN (HttpClient bricht auch das XHR/fetch ab)
readonly results = toSignal(
  this.query$.pipe(
    debounceTime(300),
    distinctUntilChanged(),
    switchMap((q) => this.api.search(q).pipe(catchError(() => of([])))),
  ),
  { initialValue: [] },
);

Wohin mit catchError

Ein Fehler beendet den Stream, in dem er auftritt. Steht catchError nach dem switchMap, schließt der erste fehlgeschlagene Request den gesamten Stream: Die Suche reagiert nicht mehr, ohne dass in der Konsole ein Fehler erscheint. Setze es in die innere Projektion. Dann wird nur dieser Request durch einen Ersatzwert ersetzt, und der äußere Stream läuft weiter.

// ✗ catchError NACH switchMap: Der erste Fehler beendet den äußeren Stream — die Suche „stirbt“
query$.pipe(
  switchMap((q) => api.search(q)),
  catchError(() => of([])),
);

// ✓ catchError IN der Projektion: Nur dieser Request schlägt fehl, der äußere Stream bleibt am Leben
query$.pipe(
  switchMap((q) => api.search(q).pipe(catchError(() => of([])))),
);

Retry mit exponentiellem Backoff

retry({ count, delay }) abonniert die Quelle nach einem Fehler erneut. Die Funktion delay erhält den Fehler und die Nummer des Versuchs und gibt ein Observable zurück, das bestimmt, wann wiederholt wird. Wird die Pause bei jedem Schritt verdoppelt (300, 600, 1200 ms …), hämmerst du nicht auf den Server ein, solange er Probleme hat. Wirf den Fehler weiter, wenn er sich nicht von selbst beheben wird (401, 404). Bei HTTP sitzt der Retry meist in einem Interceptor (siehe HttpClient & Interceptors). Für eine Verbindung, die Server-Neustarts überstehen muss — mit Jitter, damit nicht tausend Clients in derselben Sekunde zurückkommen —, siehe WebSockets.

import { retry, timer } from 'rxjs';

http.get('/api/report').pipe(
  retry({
    count: 4,                                                // höchstens 4 Retries
    delay: (error, retryCount) => {
      if (error.status === 401 || error.status === 404) throw error; // „endgültige“ Fehler nicht wiederholen
      return timer(300 * 2 ** (retryCount - 1));              // 300, 600, 1200, 2400 ms
    },
    resetOnSuccess: true,
  }),
);

share und shareReplay

Ein „kaltes“ Observable wie http.get wird bei jedem Subscribe ausgeführt. Drei Subscriber bedeuten drei Requests. share() sorgt dafür, dass gleichzeitige Subscriber sich eine einzige Ausführung teilen. shareReplay behält zusätzlich den letzten Wert für alle, die später dazukommen, und funktioniert so wie ein Cache. Die Falle ist ein einfaches shareReplay(1): Es hat refCount: false und beendet daher nie das Abonnement der Quelle. Bei einem Stream, der nie endet (Websocket, Interval), läuft die Quelle weiter, nachdem alle Subscriber gegangen sind.

Ein feiner Punkt bei refCount: true: Der Cache bleibt nur erhalten, wenn die Quelle noch abschließen konnte. Gehen alle Subscriber vor complete (zum Beispiel bei firstValueFrom, take(1) oder einer Komponente, die während des Requests zerstört wird), beendet shareReplay das Abonnement der Quelle und beginnt beim nächsten Subscriber von vorn.

// Ohne Sharing: Jedes Subscribe = eine weitere Ausführung (ein weiterer Request)
const user$ = http.get<User>('/api/me');
user$.subscribe(); user$.subscribe();                 // 2 Requests

// share(): GLEICHZEITIGE Subscriber teilen sich eine Ausführung; nach complete wird zurückgesetzt
const shared$ = http.get<User>('/api/me').pipe(share());

// shareReplay: behält den letzten Wert für SPÄTE Subscriber (Cache)
const cached$ = http.get<User>('/api/me').pipe(shareReplay({ bufferSize: 1, refCount: true }));
// refCount: true → Geht der letzte Subscriber, beendet es auch das Abonnement der Quelle.
// shareReplay(1) (refCount: false) auf einer Quelle, die nie abschließt = die Quelle läuft ewig.

Streams kombinieren

combineLatest emittiert, sobald sich irgendeine Quelle ändert, nachdem jede mindestens einmal emittiert hat. Es eignet sich für kombinierte Filter. withLatestFrom emittiert nur über die Hauptquelle und holt sich den aktuellen Wert aus den anderen („beim Klick das aktuelle Formular speichern“). forkJoin wartet, bis alle abgeschlossen sind, und emittiert einmal: Es ist ein Promise.all für Requests.

// combineLatest: emittiert bei jeder Änderung, nachdem ALLE einmal emittiert haben — kombinierte Filter
combineLatest([category$, sort$]).pipe(switchMap(([c, s]) => api.list(c, s)));

// withLatestFrom: emittiert nur über die Hauptquelle und nimmt den letzten Wert der anderen
save$.pipe(withLatestFrom(form$), concatMap(([, value]) => api.save(value)));

// forkJoin: wartet, bis ALLE abgeschlossen sind, eine einzige Emission — unabhängige parallele Requests
forkJoin({ user: api.me(), settings: api.settings() });

// zip: paart den n-ten Wert jeder Quelle — in einer UI selten nützlich

Keine vergessenen Subscriptions

Jedes manuelle subscribe in einer Komponente muss geschlossen werden. takeUntilDestroyed() erledigt das automatisch, wenn die Komponente zerstört wird. Für Werte, die im Template angezeigt werden, ist toSignal() noch einfacher: Es abonniert und beendet das Abonnement von selbst (siehe Signals ↔ RxJS).

import { takeUntilDestroyed } from '@angular/core/rxjs-interop';

export class Live {
  constructor() {
    interval(1000)
      .pipe(takeUntilDestroyed()) // in einem Injection-Kontext; sonst: takeUntilDestroyed(this.destroyRef)
      .subscribe((n) => this.tick.set(n));
  }
}
// Für Zustand, der im Template angezeigt wird, beendet toSignal(obs$) das Abonnement selbst (siehe Seite Signals ↔ RxJS).

Wann ja, wann nein

Häufige Stolperfallen

• mergeMap für die Suche. Antworten treffen in beliebiger Reihenfolge ein, und eine alte kann die neue überschreiben. Nimm switchMap.
• switchMap für Speichervorgänge. Ein auf halbem Weg abgebrochener Save wurde vom Server vielleicht schon verarbeitet, vielleicht auch nicht. Für Schreibvorgänge nimm concatMap oder exhaustMap. Mehr zu Schreibvorgängen in falscher Reihenfolge unter optimistische Updates.
• Subscribe in Subscribe. Du verlierst Abbruch und Fehlerbehandlung. Jedes „subscribe, und dann im Callback ein weiteres subscribe“ ist ein *Map-Operator.
• catchError auf der äußeren Ebene eines Streams, der am Leben bleiben muss: Es stoppt alles beim ersten Fehler.
• shareReplay(1) auf unendlichen Quellen: Speicher- und Verbindungs-Leaks. Verwende { bufferSize: 1, refCount: true }. Wie du sie findest, zeigt die Seite zu Memory Leaks.
• Retry bei 4xx: Du wiederholst endlos einen Request, der nie gelingen wird. Filtere nach Status.

Spickzettel

OPERATOR      WENN EIN NEUER WERT EINTRIFFT, WÄHREND DER ALTE LÄUFT  EINSATZ
switchMap     den alten abbrechen                                    Suche, Route-Parameter, „der letzte gewinnt“
mergeMap      parallel ausführen (optional: Concurrency)             unabhängige Operationen, Uploads
concatMap     in die Warteschlange, der Reihe nach                   geordnetes Speichern, Befehlsschlangen
exhaustMap    ignorieren                                             Submit, Login, manuelles Refresh

catchError    IN der inneren Projektion, damit der äußere Stream am Leben bleibt
retry         { count, delay: (err, n) => timer(base * 2 ** (n - 1)) } — 4xx nicht wiederholen
share()       gleichzeitige Subscriber teilen sich die Ausführung
shareReplay   { bufferSize: 1, refCount: true } — Cache + korrektes Beenden des Abonnements
combineLatest kombinierte Filter · withLatestFrom „beim Klick den aktuellen Wert nehmen“ · forkJoin „alle, einmal“
takeUntilDestroyed() / toSignal()   keine vergessenen Subscriptions