Rxjs Like A Pro

RxJS Like a Pro

This skill helps you write RxJS code that is idiomatic, composable, and free of common pitfalls. The core philosophy: keep logic in the observable chain. Every time you reach for .subscribe(), ask whether the work could instead be expressed as a transformation inside .pipe().

Reference files

For detailed examples and patterns, read the relevant reference file:

• references/loading-state-patterns.md — Deriving loading/error state in the chain, the withLoadingState custom operator, and using scan to preserve previous results across loading states. Read when working with async data fetching that needs loading indicators.
• references/massive-observable.md — How to refactor bloated new Observable() constructors into small focused pieces. Read when you see a new Observable callback longer than ~10 lines.
• references/inner-observable-chains.md — Building rich inner observable sequences with timing, delays, and animation phases. Read when composing multi-step async sequences or replacing setTimeout patterns.
• references/custom-operators.md — How to write inline and extracted custom operators with OperatorFunction. Read when extracting reusable stream logic.

The #1 Anti-pattern: Premature Subscribe

The most common RxJS mistake is subscribing too early and then doing imperative work inside the callback — tracking state in variables, calling functions with side effects, or worse, subscribing to another observable inside the callback (the "subscribe-in-subscribe" pattern).

Why this matters: when you subscribe early, you lose the power of the reactive chain. You can no longer compose, retry, cancel, debounce, or share that work. You've escaped from the declarative world into imperative spaghetti, and every new requirement (add a retry, add a timeout, combine with another stream) means more manual state management.

// ❌ Bad: subscribe-in-subscribe with manual state tracking
let currentData: Data | null = null
let loading = false

input$.subscribe((value) => {
  loading = true
  fetchData(value).subscribe((data) => {
    currentData = data
    loading = false
  })
})

// ✅ Good: everything is in the chain
const data$ = input$.pipe(switchMap((value) => fetchData(value)))

For loading state, derive it inside the chain using startWith — see references/loading-state-patterns.md.

The Massive new Observable() Antipattern

Another common antipattern is stuffing an entire program into a single new Observable(subscriber => { ... }) constructor — setting up listeners, resolving promises, subscribing to other observables, managing retry state, all in one giant callback. This is imperative code wearing an Observable costume.

The new Observable() constructor should be small and focused — a thin bridge from one non-reactive source into the reactive world. For promise-based sources, use defer(() => promise) instead. Retry logic, error handling, combining sources — all of that belongs in the operator chain.

See references/massive-observable.md for a full before/after example.

Choosing the Right Flattening Operator

Default to switchMap for most UI/request scenarios.

The inner observable doesn't have to be a single request — it can be an entire timeline of events using concat, merge, timer, delay. See references/inner-observable-chains.md for animation and timing examples.

Error Handling

Put catchError on the inner observable when you want the outer stream to keep running. Put it on the outer stream only when you truly want to replace the entire stream on error:

// ❌ Bad: catchError on outer stream kills it for good
source$.pipe(
  switchMap((value) => fetchData(value)),
  catchError((err) => of(fallback)),
)

// ✅ Good: catchError inside switchMap — outer stream survives
source$.pipe(switchMap((value) => fetchData(value).pipe(catchError((err) => of(fallback)))))

Same principle applies to retry — retry the inner operation, not the entire outer stream:

source$.pipe(
  switchMap((value) =>
    fetchData(value).pipe(
      retry({count: 3, delay: 1000}),
      catchError((err) => of(fallback)),
    ),
  ),
)

Avoiding Memory Leaks

The fewer manual subscriptions, the fewer chances to leak. In order of preference:

1. Don't subscribe at all — let the framework handle subscription lifecycle where possible
2. Use operators that complete naturally — first(), take(n), takeUntil(destroy$)
3. Use takeUntil with a notifier:

const destroy$ = new Subject<void>();
someObservable$.pipe(
  takeUntil(destroy$),
).subscribe(value => /* ... */);

// In teardown: destroy$.next(); destroy$.complete();

takeUntil must be the last operator in the pipe. Operators after it (especially flattening operators) can create inner subscriptions that takeUntil doesn't know about, causing leaks.

4. Compose into a single subscription — if you have multiple independent streams with side effects, merge them into one and subscribe once.

Hot vs Cold

• Cold observables (new Observable(...), of(), HTTP requests) create a new execution per subscriber
• Hot observables (Subject, fromEvent) share a single execution

Share cold observables with shareReplay({ bufferSize: 1, refCount: true }). Always use refCount: true — without it, the source subscription stays alive after all subscribers unsubscribe (memory leak).