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).

Deriving State Reactively

Instead of mutable variables updated from multiple subscriptions, derive state from streams:

// ❌ Bad: mutable state, inconsistent windows
let items: Item[] = []
let filter = ''
items$.subscribe((i) => {
  items = i
  recompute()
})
filter$.subscribe((f) => {
  filter = f
  recompute()
})

// ✅ Good: always consistent
const filteredItems$ = combineLatest([items$, filter$]).pipe(
  map(([items, filter]) => items.filter((item) => item.name.includes(filter))),
)

`combineLatest` vs `withLatestFrom`: combineLatest emits when _any_ input emits (all inputs drive output). withLatestFrom emits only when the _source_ emits (one driver, others are context).

`startWith`: combineLatest won't emit until every input has emitted at least once. Use startWith to provide initial values and unblock the stream.

Subjects: Use Sparingly

Subject, BehaviorSubject, ReplaySubject are escape hatches for bridging imperative and reactive code. Appropriate for event buses and bridging callbacks. _Not_ appropriate as general-purpose state containers — if you're calling .next() in multiple places to keep a Subject in sync, use a derived stream instead.

Custom Operators

Don't be afraid to write them — they're just functions with the signature (source: Observable<A>) => Observable<B>. Extract repeated .pipe() chains into named operators with OperatorFunction<In, Out>. See references/custom-operators.md for inline and extracted examples.

Side Effects Belong in tap, Not in subscribe

A good rule of thumb: .subscribe() should have no arguments. All side effects — logging, updating the DOM, writing to a database, sending analytics — belong in tap inside the chain. The .subscribe() at the end just activates the stream.

// ❌ Bad: side effects crammed into subscribe
source$.pipe(switchMap((value) => fetchData(value))).subscribe(
  (data) => {
    updateUI(data)
    logAnalytics('data_loaded', data)
    cache.set(data)
  },
  (err) => showError(err),
)

// ✅ Good: side effects in tap, subscribe just activates
source$
  .pipe(
    switchMap((value) => fetchData(value)),
    tap((data) => updateUI(data)),
    tap((data) => logAnalytics('data_loaded', data)),
    tap((data) => cache.set(data)),
    tap({error: (err) => showError(err)}),
  )
  .subscribe()

Why this matters: when side effects are in the chain, they're composable. You can add, remove, or reorder them. You can put a filter between them. You can share the stream and have different subscribers without duplicating side-effect logic. When everything is stuffed into .subscribe(), you've lost all of that.

tap also accepts an observer object with lifecycle hooks — particularly useful for debugging:

source$.pipe(
  tap({
    subscribe: () => console.log('subscribed!'),
    next: (value) => console.log('value:', value),
    error: (err) => console.log('error:', err),
    complete: () => console.log('complete'),
    unsubscribe: () => console.log('unsubscribed'),
    finalize: () => console.log('finalized (complete or unsubscribe)'),
  }),
)

The subscribe hook is especially handy for debugging "why isn't my stream emitting?" — it confirms whether anything is actually subscribing.

Avoid Unnecessary Promise Conversion

firstValueFrom/lastValueFrom are appropriate for one-shot interop with promise-based APIs. They're a code smell when used inside subscribe callbacks to avoid learning the reactive approach — that work belongs in the chain with switchMap.

Quick Reference: Common Refactoring Patterns

Custom Operators

Custom operators are just functions that take an observable and return an observable. They're how you make reusable, composable pieces of stream logic — and they're simpler than they look.

Inline custom operators

The simplest form is a function you write right in the pipe. Any time you find yourself doing the same multi-step .pipe() chain in several places, that's a candidate:

// An inline operator is just a function: Observable<In> → Observable<Out>
const results$ = searchInput$.pipe(
  // inline operator — debounce, deduplicate, and skip empty
  (source) =>
    source.pipe(
      debounceTime(300),
      distinctUntilChanged(),
      filter((query) => query.length > 0),
    ),
  switchMap((query) => apiService.search(query)),
)

This works because .pipe() accepts any function with the signature (source: Observable<A>) => Observable<B>.

Extracting into a reusable operator

When you use the same inline operator in multiple places, extract it into a named function. Use OperatorFunction<In, Out> as the return type so it slots cleanly into any .pipe() chain:

import {OperatorFunction, Observable} from 'rxjs'
import {debounceTime, distinctUntilChanged, filter} from 'rxjs/operators'

interface StabilizeOptions {
  debounce?: number
  minLength?: number
}

function stabilizeInput(options: StabilizeOptions = {}): OperatorFunction<string, string> {
  const {debounce = 300, minLength = 1} = options
  return (source: Observable<string>) =>
    source.pipe(
      debounceTime(debounce),
      distinctUntilChanged(),
      filter((query) => query.length >= minLength),
    )
}

// Now reusable across any text input stream:
const search$ = searchInput$.pipe(
  stabilizeInput({debounce: 200}),
  switchMap((query) => apiService.search(query)),
)

const autocomplete$ = nameInput$.pipe(
  stabilizeInput({debounce: 500, minLength: 2}),
  switchMap((name) => apiService.suggest(name)),
)

The pattern is always the same:

1. Write a function that accepts configuration (if any) and returns OperatorFunction<In, Out> 2. The returned function takes a source observable and pipes operators onto it 3. Use it in .pipe() just like any built-in operator

Custom operators are a sign of mature RxJS code. They encapsulate domain-specific stream logic (debounce policies, retry strategies, polling intervals, auth token refresh) into tested, named, reusable pieces. When you see repeated .pipe() chains or complex inline logic, extract an operator.

Composing Inner Observable Chains

One of the most powerful — and underused — patterns in RxJS is building rich inner observable chains inside flattening operators. People often think of switchMap as "map a value to a single request", but the inner observable can be an entire sequence of events with its own timing, ordering, and lifecycle.

Delayed actions with inner observables

Instead of using setTimeout in a subscribe callback, compose timing directly:

// ❌ Bad: imperative timeout inside subscribe
click$.subscribe(() => {
  showTooltip()
  setTimeout(() => hideTooltip(), 3000)
})

// ✅ Good: timing is part of the chain
click$
  .pipe(switchMap(() => concat(of('show'), timer(3000).pipe(map(() => 'hide')))))
  .subscribe((action) => {
    action === 'show' ? showTooltip() : hideTooltip()
  })

The reactive version gets cancellation for free — if the user clicks again, switchMap tears down the previous timer and starts fresh. The imperative version would need manual clearTimeout tracking.

Animation sequences

Inner observable chains are perfect for multi-phase sequences. Each phase is an observable, and concat plays them in order:

// A notification that fades in, stays visible, then fades out
function showNotification(message: string) {
  return concat(
    // Phase 1: fade in over 300ms
    animateOpacity(0, 1, 300).pipe(map((opacity) => ({message, opacity, visible: true}))),
    // Phase 2: stay visible for 3 seconds
    timer(3000).pipe(map(() => ({message, opacity: 1, visible: true}))),
    // Phase 3: fade out over 300ms
    animateOpacity(1, 0, 300).pipe(map((opacity) => ({message, opacity, visible: opacity > 0}))),
  )
}

// Using it — each new notification cancels the previous one mid-animation
notification$
  .pipe(switchMap((message) => showNotification(message)))
  .subscribe((state) => render(state))

The key insight: the inner observable returned by a flattening operator doesn't have to be a single HTTP request — it can be an entire timeline of events. concat, merge, timer, interval, delay — these are all tools for composing rich inner sequences. And the outer flattening operator (switchMap, concatMap, etc.) handles the lifecycle of the whole sequence as a unit.

Loading State Patterns

Deriving loading state in the chain

Instead of tracking loading/error in mutable variables, derive it inside the switchMap using startWith:

const dataWithLoadingState$ = input$.pipe(
  switchMap((value) =>
    fetchData(value).pipe(
      map((data) => ({loading: false, data})),
      catchError((error) => of({loading: false, error})),
      startWith({loading: true}),
    ),
  ),
)

Extract into a reusable custom operator

import {OperatorFunction, Observable, of} from 'rxjs'
import {switchMap, map, catchError, startWith} from 'rxjs/operators'

type LoadingState<T> =
  | {loading: true}
  | {loading: false; data: T}
  | {loading: false; error: unknown}

function withLoadingState<T, R>(
  project: (value: T) => Observable<R>,
): OperatorFunction<T, LoadingState<R>> {
  return (source) =>
    source.pipe(
      switchMap((value) =>
        project(value).pipe(
          map((data) => ({loading: false, data}) as const),
          catchError((error) => of({loading: false, error} as const)),
          startWith({loading: true} as const),
        ),
      ),
    )
}

// Now any stream can use it:
const results$ = searchInput$.pipe(withLoadingState((query) => apiService.search(query)))

Once a pattern is in an operator, it's tested once and reusable everywhere. The loading/error/data lifecycle is guaranteed consistent across every stream that uses it.

Preserving previous results across loading states

When a new input arrives, switchMap cancels the previous inner observable and starts fresh with { loading: true }. This means the UI loses the previous results during the loading phase — the user sees a blank or spinner instead of the stale-but-still-useful data they were just looking at.

Use scan to carry forward previous results while new ones are loading:

const results$ = searchInput$.pipe(
  withLoadingState((query) => apiService.search(query)),
  scan((previous, current) => {
    if (current.loading) {
      // Keep showing previous data while loading
      return {...current, data: 'data' in previous ? previous.data : undefined}
    }
    if ('error' in current) {
      // On error, keep the previous data so the UI doesn't blank out,
      // but surface the error so it can be displayed
      return {...current, data: 'data' in previous ? previous.data : undefined}
    }
    return current
  }),
)

Now the UI can:

• Show a loading indicator _and_ keep displaying previous results until new ones arrive
• On error, show the error message while still displaying the last successful results
• On success, replace everything with the fresh data

This avoids the jarring pattern where a transient network error wipes out perfectly good data the user was just looking at.

The same scan pattern works for any situation where you want to "remember" something across emissions — accumulating a list, tracking a running total, or preserving context that would otherwise be lost when the stream moves to its next state.

The Massive new Observable() Antipattern

A common antipattern is stuffing an entire program into a single new Observable(subscriber => { ... }) constructor. These tend to grow into massive imperative blocks — setting up event listeners, resolving promises, subscribing to other observables, managing state — all inside one giant callback:

// ❌ Bad: an entire application crammed into a single Observable constructor
const data$ = new Observable((subscriber) => {
  let retryCount = 0
  const controller = new AbortController()

  async function doFetch() {
    try {
      const response = await fetch('/api/data', {signal: controller.signal})
      const json = await response.json()
      subscriber.next(json)

      // Now set up an EventSource for live updates...
      const es = new EventSource('/api/updates')
      es.onmessage = (event) => {
        const update = JSON.parse(event.data)
        subscriber.next(update)
      }
      es.onerror = () => {
        es.close()
        if (retryCount < 3) {
          retryCount++
          setTimeout(doFetch, 1000)
        } else {
          subscriber.error(new Error('EventSource failed'))
        }
      }
      subscriber.add(() => es.close())
    } catch (err) {
      subscriber.error(err)
    }
  }
  doFetch()

  return () => controller.abort()
})

This is imperative code wearing an Observable costume. It manually tracks retry count, manages cleanup, handles errors with try/catch — all things RxJS has operators for.

The fix: small constructors + operator composition

The new Observable() constructor should be small and focused — the bridge from a _single_ non-reactive source into the reactive world:

// ✅ Good: small focused observables composed together
// Step 1: small Observable constructor — bridges EventSource and nothing else
function fromEventSource(url: string) {
  return new Observable<MessageEvent>((subscriber) => {
    const es = new EventSource(url)
    es.onmessage = (event) => subscriber.next(event)
    es.onerror = () => subscriber.error(new Error('EventSource connection failed'))
    return () => es.close()
  })
}

// Step 2: compose everything with operators
const updates$ = fromEventSource('/api/updates').pipe(
  retry({count: 3, delay: (_, attempt) => timer(attempt * 1000)}),
  map((event) => JSON.parse(event.data)),
)

const initialSnapshot$ = new Observable<Response>((subscriber) => {
  const controller = new AbortController()
  fetch('/api/data', {signal: controller.signal}).then(
    (response) => {
      subscriber.next(response)
      subscriber.complete()
    },
    (err) => subscriber.error(err),
  )
  return () => controller.abort()
}).pipe(mergeMap((response) => response.json()))

const data$ = merge(initialSnapshot$, updates$)

Notice how defer(() => somePromise) often replaces new Observable entirely for promise-based sources. defer is lazy — it won't call the function until someone subscribes, and each subscriber gets a fresh execution. The JSON parsing moves to a map in the chain — keeping the fromEventSource bridge generic and reusable.

Rules of thumb

• new Observable() should be a few lines that bridge _one_ non-reactive source (a DOM event, an

EventSource, a callback-based API) — the thinnest possible adapter

• defer(() => promise) replaces new Observable for anything promise/async-based
• Retry logic, error handling, combining sources, timing — all of that belongs in the operator chain, not

inside the constructor

• If your new Observable callback is longer than ~10 lines, it's probably doing too much