Angular RxJS patterns

Why the HTTP call never fired

You call http.get('/api/profile'), no request leaves the browser, no error. The cause is that an Observable is lazy. Creating it only describes the work; the producer runs when someone subscribes. Angular's HttpClient is explicit that no request is dispatched until you subscribe, so a get you never subscribe to is a request you never sent.

This is the first thing that trips people coming from promises. A promise is eager: fetch(url) fires the moment you call it, whether or not you await it. An Observable does nothing until subscription, runs the producer again for each new subscriber, and can push more than one value over time. Three practical differences fall out of that:

• Lazy vs eager: subscribe to start, and subscribing twice runs an HTTP request twice.
• Many values vs one: valueChanges, WebSocket feeds, and router events keep emitting; a promise resolves once.
• Cancellable vs not: unsubscribing from an HttpClient call aborts the in-flight request. A promise has no cancel.

You rarely call .subscribe() by hand in a component. Prefer the async pipe or toSignal (covered below), which subscribe for you and clean up on their own.

Stop nesting subscribes

The most common RxJS bug in Angular interviews is a subscribe inside a subscribe:

// bug: a race waiting to happen
this.searchControl.valueChanges.subscribe(term => {
  this.http.get<Result[]>('/api/search?q=' + term)
    .subscribe(res => this.results = res);
});

Each keystroke starts its own request and nothing cancels the last one. Requests run in parallel and can resolve out of order, so a slow response for "re" can land after a fast one for "red" and overwrite it. The results box then shows matches for a word the user already changed.

The fix is a flattening operator: it takes each outer value, maps it to an inner Observable, and subscribes to that inner stream for you (no nesting). For a typeahead you want switchMap, which keeps only the newest inner subscription. When a new term arrives it unsubscribes the previous inner request, and because HttpClient aborts on unsubscribe, the stale response is cancelled at the network layer.

results$ = this.searchControl.valueChanges.pipe(
  debounceTime(250),          // wait for a pause in typing
  distinctUntilChanged(),     // skip if the term did not change
  switchMap(term => this.http.get<Result[]>('/api/search?q=' + term)),
);

The four flattening operators differ only in what they do with a new outer value while an inner one is still running:

Search wants switchMap because only the latest term matters. A submit button wants exhaustMap so a double-click cannot fire the request twice. Ordered writes (save step 1, then step 2) want concatMap so they never overtake each other. mergeMap runs everything at once with no ordering, which is right for independent fire-and-forget work and wrong for anything where order or cancellation matters.

Clean up without the boilerplate

A subscription you make by hand keeps running until you tear it down or the source completes. That is where leaks come from: a valueChanges or route-params subscription made in ngOnInit never completes on its own, so it survives the component and keeps firing after the view is gone.

Wrong "every subscribe leaks unless you unsubscribe." A subscription to a source that completes cleans itself up. A single http.get emits once and completes, so its subscription closes on its own. Right the leak is long-lived sources that never complete (form valueChanges, a Subject, interval, router events). Those are the ones you must bound.

The best fix is to not subscribe by hand at all. Render the stream with the async pipe and Angular subscribes when the view renders and unsubscribes when it is destroyed:

<ul>
  <li *ngFor="let r of results$ | async">{{ r.title }}</li>
</ul>

The async pipe also marks the component for check on each emission, so it updates an OnPush view (the default for new components in Angular v22) with no manual wiring. When you genuinely need an imperative subscribe (a side effect, not a value you render), reach for takeUntilDestroyed():

constructor() {
  this.events$
    .pipe(takeUntilDestroyed())   // completes the stream when this component is destroyed
    .subscribe(e => this.log(e));
}

Called with no argument it must run in an injection context (a constructor or field initializer), where it finds the current DestroyRef itself. If you call it later, in a method, pass a DestroyRef you injected: takeUntilDestroyed(this.destroyRef). Either way it replaces the old takeUntil(this.destroy$) pattern with its manual Subject and ngOnDestroy.

switchMap cancels, which is sometimes the bug

switchMap is the default answer for reads because cancelling the previous inner request is what you want: the latest term wins, the stale one aborts. Put it on a write and that same cancellation loses data. A user clicks save, then quickly edits and clicks save again; switchMap unsubscribes the first request mid-flight, and if the server had already begun the write you now have a half-applied change with no client-side record it happened.

Match the operator to the semantics of the action:

• switchMap for reads where only the latest matters: typeahead, a details panel that reloads when the selected id changes.
• exhaustMap for idempotent triggers you want to debounce by ignoring: a submit button, a refresh button. While the request is in flight, extra clicks are dropped, so a double-click cannot double-submit.
• concatMap for writes that must all happen and keep order: queue each request and run it only after the previous inner Observable completes (it is mergeMap with concurrency 1). Nothing overtakes, nothing is dropped.
• mergeMap for independent work with no ordering constraint, and only where unbounded concurrency is acceptable or you cap it with the concurrency argument.

Wrong "switchMap is the safe default everywhere." On a mutation it silently discards in-flight writes. The safe default for a write is concatMap (keep them all, in order) or exhaustMap (drop the duplicates), never the operator that cancels.

Cold vs hot, and why one GET became five

A cold Observable runs its producer once per subscriber. HttpClient is cold, so five components doing config$ | async in their templates fire five identical GETs. Multicasting fixes this: one subscription to the source, its values pushed to many observers through a Subject. share() does exactly that with reference counting: it subscribes to the source on the first subscriber and unsubscribes when the count drops to zero.

For a value that arrives once and should be remembered, you want shareReplay(1), which multicasts through a ReplaySubject and replays the last value to late subscribers. That is the right tool for a config or a current-user request. It has one trap that interviewers probe:

// one-shot GET: refCount false is harmless here, the buffer is held for the app's lifetime
config$ = this.http.get<Config>('/config').pipe(shareReplay(1));

// an infinite source leaks with the default; refCount:true tears down at zero subscribers
live$ = this.socket.messages$.pipe(shareReplay({ bufferSize: 1, refCount: true }));

By default shareReplay uses refCount: false: after the source emits it stays subscribed even when every consumer has unsubscribed, holding the buffer for the app's lifetime. On a one-shot HTTP request that is harmless and even wanted. On a source that never completes it is a leak, because the source keeps running with nobody listening.

Wrong "a default shareReplay caches the error and replays it to every future subscriber forever, so you must put retry before it." That was RxJS 6. Current shareReplay defaults to resetOnError: true: an error resets the internal ReplaySubject, so it is not cached. A later subscriber, or a retry in either position, re-runs the source. Only a successfully completed value is held for the app's lifetime, so the real footgun is serving stale success data (plus the refCount: false leak on infinite sources), not error poisoning.

Where catchError goes

catchError catches an error from anywhere upstream in its pipe and returns a replacement Observable to continue with. Placement decides what survives the failure. Inside the inner request it contains the blast radius to one call; on the outer stream it terminates the whole thing.

results$ = this.term$.pipe(
  switchMap(term => this.http.get<Result[]>('/api/search?q=' + term).pipe(
    catchError(() => of([])),   // one failed search -> empty results, stream lives on
  )),
);

Wrong "one catchError at the end of the pipe handles all errors." An Observable terminates on error. If the error propagates up to a catchError on the outer term$ stream, that stream has already ended, so the next keystroke produces nothing and the search box is dead until reload. Catch at the inner level for per-request recovery; reserve an outer catchError for genuinely fatal cases.

retry resubscribes to the source on error, up to count times. Bare retry(3) hammers a failing server instantly; give it a backoff with the delay option, and resetOnSuccess if you want the counter to reset after a good emission on a long-lived stream.

combineLatest and forkJoin fire at different times

Both combine streams, and picking the wrong one gives you a component that renders nothing.

forkJoin waits for every input to complete, then emits one array of their last values. It is the RxJS "wait for all of these to finish", right for a handful of one-shot HTTP calls on load. If any input never completes it never emits, and if any input errors the whole thing errors and unsubscribes the rest.

combineLatest emits whenever any input emits, but only after every input has emitted at least once. That "at least once" is the gotcha: pair a stream that has fired with one that has not (a Subject nobody has pushed to yet, a form control the user has not touched) and combineLatest sits silent, because it is still waiting for the quiet one. Seed such inputs with startWith so they emit immediately.

Wrong "combineLatest is like forkJoin but live." forkJoin waits for completion and emits once at the end; combineLatest emits on every change and never needs its inputs to complete. Using forkJoin on inputs that never complete (a valueChanges, a router stream) means it never emits at all.

Signals interop: which way to cross the boundary

Angular's @angular/core/rxjs-interop package is the bridge between RxJS and signals. Use it to move state across the boundary, not to replace one with the other.

toSignal(obs$) subscribes to an Observable and exposes its latest value as a signal, and unsubscribes automatically when the enclosing component or service is destroyed. It is the signal-world equivalent of the async pipe and works anywhere, not just templates. Two options carry the weight: initialValue sets what the signal reads before the first emission, and requireSync: true asserts the source emits synchronously so no initial value is needed (it throws if the source does not). Reading the signal after the source errored re-throws that error, and you should call toSignal once per Observable and reuse the result rather than calling it repeatedly.

// Observable -> signal, subscription owned by the component
user = toSignal(this.userService.current$, { initialValue: null });

toObservable(sig) goes the other way: it watches a signal through an effect and emits when the signal's value settles, so it needs an injection context (or an explicit Injector). Reach for it when a signal input needs to drive an RxJS pipeline, for instance debouncing a signal-based search box:

query = input('');   // signal input
results$ = toObservable(this.query).pipe(
  debounceTime(250),
  switchMap(q => this.http.get<Result[]>('/api/search?q=' + q)),
);

The rule of thumb: keep async orchestration (HTTP, debouncing, cancellation, retries, merging event streams) in RxJS, keep synchronous derived state in signals, and cross the boundary once at the edge with toSignal or toObservable. The signals primitive itself, change detection, and when a computed beats an effect live in the change-detection-and-signals module; here the boundary is the point.