rxjs-consistency

RxJS — consistent, modern idioms

RxJS is stable at v7, but training data spans every era: v5 patch imports (rxjs/add/operator/map), method chaining, toPromise(), and v6/7 pipeable style all surface in generated code — plus the timeless sins of nested subscribes and leaked subscriptions. This skill pins one canonical idiom set — RxJS 7 pipeable style with explicit subscription lifetimes and deliberate flattening.

Canonical idioms — always X, never Y

Always	Never	Why
source$.pipe(map(f), filter(g))	source$.map(f).filter(g) / rxjs/add/operator/* imports	Prototype patching died with v5; only pipeable operators exist.
await firstValueFrom(req$) / lastValueFrom	await req$.toPromise()	toPromise is deprecated (removed in v8) and resolves undefined on empty — the named functions make the choice explicit and error on empty.
one flattening operator: switchMap/mergeMap/concatMap/exhaustMap	subscribe inside subscribe	Nesting orphans the inner subscription (leaks, no cancellation, race conditions); flattening manages it.
takeUntil(this.destroy$) as the last state-affecting operator (or takeUntilDestroyed in Angular 16+)	fire-and-forget .subscribe() on long-lived streams	Infinite streams outlive the component — the classic RxJS memory leak.
subscribe({ next, error, complete }) observer object	subscribe(nextFn, errFn, doneFn) multi-callback	The multi-argument signature is deprecated.
catchError placed on the inner observable	one catchError at the end of a long-lived stream	An error reaching the outer stream completes it permanently — the "my polling stopped forever" bug.
shareReplay({ bufferSize: 1, refCount: true }) (or share() with reset options)	bare shareReplay(1) on infinite sources	Without refCount the source keeps running with zero subscribers — a leak by default.
naming convention foo$ and exposing asObservable()	exposing raw Subjects from services	Callers shouldn't be able to next() into your state.
Angular: async pipe in templates	manual subscribe + class property + change detection	The pipe subscribes, unsubscribes, and triggers CD for you.

The flattening decision table (memorize this):

Need	Operator
Latest wins, cancel the previous (typeahead, route params → load)	switchMap
Run all concurrently, order doesn't matter (independent writes, logging)	mergeMap
Strict order, one at a time (queued writes)	concatMap
Ignore triggers while busy (submit button, login)	exhaustMap

House style for a search box:

results$ = this.query$.pipe(
  debounceTime(300),
  distinctUntilChanged(),
  switchMap((q) =>
    this.api.search(q).pipe(
      catchError(() => of([] as Result[])),   // inner: keep the outer stream alive
    ),
  ),
  takeUntilDestroyed(),                        // Angular 16+; else takeUntil(destroy$)
);

Pitfalls that produce silently wrong behavior

• Wrong flattener: mergeMap on a typeahead lets a slow stale response overwrite a fresh one; switchMap on saves cancels in-flight writes. Match the table, every time.
• Cold observables re-execute per subscriber: two subscribes (or two async pipes) on one HTTP observable = two requests. Multicast with shareReplay({bufferSize: 1, refCount: true}) when sharing is intended.
• takeUntil placed before a multicasting/flattening operator doesn't stop the inner work; keep it last (before final side-effect-free ops).
• BehaviorSubject vs Subject vs ReplaySubject: late subscribers get current value / nothing / a replay buffer respectively — choosing wrong silently drops or duplicates state.
• firstValueFrom on a stream that may complete empty rejects with EmptyError; pass { defaultValue: ... } when empty is legitimate.
• Side effects in map instead of tap make streams non-idempotent under retry/replay.
• finalize vs complete: cleanup that must run on unsubscribe-or-error belongs in finalize; complete callbacks never fire on takeUntil-style teardown via error.

Version notes

Target RxJS 7. Breaking line v5→v6: pipeable operators, rxjs/operators imports. v6→v7: toPromise deprecated (firstValueFrom/lastValueFrom added), stricter types, share/shareReplay reset options, most operators importable from rxjs root. v8 removes the deprecated signatures — write v7 code that is already v8-clean.

Workflow

1. Model the source: user events → Subject; state → BehaviorSubject; external pushes → fromEvent/from; one-shot async → defer/from(promise).
2. Compose with pipeable operators; pick the flattener from the table for every inner async step; catchError on the inner observable with an explicit fallback.
3. Decide each stream's lifetime: who unsubscribes, where (takeUntil/ takeUntilDestroyed/async pipe/first()-style self-completing).
4. Multicast deliberately (shareReplay with refCount) where multiple consumers share one execution; otherwise accept cold semantics knowingly.
5. When reviewing, hunt: nested subscribes, toPromise, patch imports, bare .subscribe() on infinite streams, outer-level catchError on long-lived streams, multi-callback subscribe.

For the fuller operator catalogs (combination, error/retry, multicasting semantics) and Subject/scheduler reference, read references/rxjs-patterns.md.