Advanced RxJS: higher-order operators

Signals have taken over the state, but events over time remain the territory of RxJS: keystrokes, clicks, overlapping requests, retries, websockets. The most important part, and the one most often got wrong, is the higher-order operators. Each value starts a new Observable, usually an HTTP request. The real question is: what do you do when the next value arrives while the previous request is still running?

There are four answers: cancel, run in parallel, queue, or ignore. Each has an operator. If you choose wrongly, the result is a race condition bug: old results overwriting the new ones, duplicate orders, saves applied in the wrong order.

The problem: Observable of Observables

A map that returns a request produces an Observable<Observable<T>>: a stream of requests that nobody has subscribed to. The *Map operators subscribe to the inner streams themselves and flatten them into a single stream of results. The difference between them is only the overlap policy.

// "For each value, start an Observable" — the question is what you do when the next value arrives
// while the previous one is still running.
clicks$.pipe(map((id) => http.get(`/api/items/${id}`)));       // ✗ Observable<Observable<Item>>
clicks$.pipe(switchMap((id) => http.get(`/api/items/${id}`))); // ✓ Observable<Item> — plus a policy

The four operators

// Typeahead: only the last request matters; the old ones are CANCELLED (HttpClient also cancels the XHR/fetch)
readonly results = toSignal(
  this.query$.pipe(
    debounceTime(300),
    distinctUntilChanged(),
    switchMap((q) => this.api.search(q).pipe(catchError(() => of([])))),
  ),
  { initialValue: [] },
);

Where to put catchError

An error completes the stream in which it occurs. If catchError sits after switchMap, the first failed request closes the whole stream: the search stops responding, with no error in the console. Put it inside the inner projection. Then only that request is replaced with a fallback value, and the outer stream carries on.

// ✗ catchError AFTER switchMap: the first error completes the outer stream — the search "dies"
query$.pipe(
  switchMap((q) => api.search(q)),
  catchError(() => of([])),
);

// ✓ catchError INSIDE the projection: only that request fails, the outer stream stays alive
query$.pipe(
  switchMap((q) => api.search(q).pipe(catchError(() => of([])))),
);

Retry with exponential backoff

retry({ count, delay }) resubscribes to the source after an error. The delay function receives the error and the retry number, and returns an Observable that says when to retry. Doubling the pause at each step (300, 600, 1200 ms…) does not hammer the server while it has problems. Rethrow the error for codes that will not fix themselves (401, 404). For HTTP, retry usually lives in an interceptor (see HttpClient & interceptors). For a connection that must survive server restarts — with jitter, so a thousand clients don’t come back in the same second — see WebSockets.

import { retry, timer } from 'rxjs';

http.get('/api/report').pipe(
  retry({
    count: 4,                                                // at most 4 retries
    delay: (error, retryCount) => {
      if (error.status === 401 || error.status === 404) throw error; // do not retry "definitive" errors
      return timer(300 * 2 ** (retryCount - 1));              // 300, 600, 1200, 2400 ms
    },
    resetOnSuccess: true,
  }),
);

share and shareReplay

A “cold” Observable, such as http.get, runs on every subscribe. Three subscribers mean three requests. share() makes simultaneous subscribers share a single execution. shareReplay also keeps the last value for those who arrive later, so it works like a cache. The trap is a plain shareReplay(1): it has refCount: false, so it never unsubscribes from the source. On a stream that never ends (a websocket, an interval), the source keeps running after all the subscribers have left.

A subtle detail with refCount: true: the cache is kept only if the source managed to finish. If all subscribers leave before complete (for example firstValueFrom, take(1), or a component destroyed during the request), shareReplay unsubscribes from the source and starts over with the next subscriber.

// Without sharing: each subscribe = one more execution (one more request)
const user$ = http.get<User>('/api/me');
user$.subscribe(); user$.subscribe();                 // 2 requests

// share(): SIMULTANEOUS subscribers share one execution; after complete it resets
const shared$ = http.get<User>('/api/me').pipe(share());

// shareReplay: keeps the last value for LATE subscribers (cache)
const cached$ = http.get<User>('/api/me').pipe(shareReplay({ bufferSize: 1, refCount: true }));
// refCount: true → when the last subscriber leaves, it also unsubscribes from the source.
// shareReplay(1) (refCount: false) on a source that never completes = the source runs forever.

Combining streams

combineLatest emits whenever any source changes, once each has emitted at least once. It suits combined filters. withLatestFrom emits only from the main source and takes the current value from the others (“on click, save the current form”). forkJoin waits for all of them to complete and emits once: it is a Promise.all for requests.

// combineLatest: emits on every change, after ALL of them have emitted once — combined filters
combineLatest([category$, sort$]).pipe(switchMap(([c, s]) => api.list(c, s)));

// withLatestFrom: emits only from the main source, taking the latest value from the others
save$.pipe(withLatestFrom(form$), concatMap(([, value]) => api.save(value)));

// forkJoin: waits for ALL of them to complete, a single emission — independent parallel requests
forkJoin({ user: api.me(), settings: api.settings() });

// zip: pairs the n-th value of each — rarely useful in a UI

No forgotten subscriptions

Every manual subscribe in a component must be closed. takeUntilDestroyed() does it automatically when the component is destroyed. For values shown in the template, toSignal() is even simpler: it subscribes and unsubscribes by itself (see Signals ↔ RxJS).

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

export class Live {
  constructor() {
    interval(1000)
      .pipe(takeUntilDestroyed()) // in an injection context; otherwise: takeUntilDestroyed(this.destroyRef)
      .subscribe((n) => this.tick.set(n));
  }
}
// For state shown in the template, toSignal(obs$) unsubscribes by itself (see the Signals ↔ RxJS page).

When YES, when NO

Common pitfalls

• mergeMap for search. Responses arrive in any order, and an old one can overwrite the new one. Use switchMap.
• switchMap for saves. A save cancelled halfway may already have been processed by the server, or may not. For writes, use concatMap or exhaustMap. More on writes that arrive out of order in optimistic updates.
• Subscribe inside subscribe. You lose cancellation and error handling. Any “subscribe, then another subscribe in the callback” is a *Map operator.
• catchError at the outer level of a stream that must stay alive: it stops everything at the first error.
• shareReplay(1) on infinite sources: memory and connection leaks. Use { bufferSize: 1, refCount: true }. How to find them is on the memory leaks page.
• Retry on 4xx: you retry a request forever that will never succeed. Filter by status.

Cheat sheet

OPERATOR      WHEN A NEW VALUE ARRIVES WHILE THE OLD ONE RUNS  USE FOR
switchMap     cancel the old one                               search, route parameters, "the last one wins"
mergeMap      run them in parallel (optional: concurrency)     independent operations, uploads
concatMap     queue it, in order                               ordered saves, command queues
exhaustMap    ignore it                                        submit, login, manual refresh

catchError    INSIDE the inner projection, so the outer stream stays alive
retry         { count, delay: (err, n) => timer(base * 2 ** (n - 1)) } — don't retry 4xx
share()       simultaneous subscribers share the execution
shareReplay   { bufferSize: 1, refCount: true } — cache + correct unsubscription
combineLatest combined filters · withLatestFrom "on click, take the current value" · forkJoin "all of them, once"
takeUntilDestroyed() / toSignal()   no forgotten subscriptions