RxJS Reactive Programming Patterns

Quick Guide: RxJS composes async work as streams. Observables are lazy — nothing runs until subscribe() — and operators inside pipe() describe a pipeline rather than execute one. Three decisions carry most of the weight: which flattening operator (switchMap cancels, mergeMap parallelises, concatMap queues, exhaustMap ignores), where catchError sits (inside the inner pipe, or an error kills the outer stream permanently), and how the subscription ends. Suffix observable variables with $.

Detailed Resources:

• examples/core.md — creation, lazy execution with defer, transformation, error recovery, Promise interop
• examples/higher-order.md — switchMap, mergeMap, concatMap, exhaustMap worked through
• examples/subjects.md — the four Subject types, multicasting, shareReplay
• examples/combination.md — combineLatest, forkJoin, merge, concat, race
• examples/memory-leaks.md — the takeUntil pattern, subscription collections, self-completing streams
• reference.md — operator tables, scenario-to-operator lookup, deprecations

---

Which path applies

Cleanup is the branch, and it is decided by whether the source ends on its own.

• The source completes by itself — of, from, a wrapped Promise, timer(n), a single HTTP response. Nothing to unsubscribe; go straight to examples/core.md.
• The source runs until stopped — fromEvent, interval, any Subject, a WebSocket. Every subscription needs an ending, and the shapes are in examples/memory-leaks.md.
• You need one value out of a stream — firstValueFrom / lastValueFrom bridge to await and settle the lifetime for you. In examples/core.md.

---

<critical_requirements>

Before writing RxJS code

Give every long-lived subscription an ending — takeUntil with a destroy Subject, a Subscription collection, or a self-completing operator such as take or first. A fromEvent or interval subscription with no ending outlives whatever created it, holding its closure alive.

Put takeUntil last in the pipe. Operators only see what precedes them, so a takeUntil above a switchMap ends the outer source and leaves the inner Observable running — a leak that looks like correct cleanup code.

Put catchError inside the inner pipe, not the outer one. An error propagates to the subscriber and terminates the chain for good; recovering inside the inner Observable keeps the outer stream alive for the next value.

Pick the flattening operator from the concurrency you want — cancel previous (switchMap), run in parallel (mergeMap), queue (concatMap), ignore while busy (exhaustMap). Defaulting to mergeMap for a search box is how stale responses overwrite fresh ones.

</critical_requirements>

---

Auto-detection: RxJS, rxjs, Observable, Subject, BehaviorSubject, ReplaySubject, AsyncSubject, subscribe, pipe, switchMap, mergeMap, concatMap, exhaustMap, combineLatest, forkJoin, withLatestFrom, fromEvent, defer, interval, timer, catchError, retry, debounceTime, throttleTime, distinctUntilChanged, takeUntil, shareReplay, firstValueFrom, lastValueFrom, Subscription

Applies to:

• Event streams that need composition — debounce, throttle, buffer, distinct
• Coordinating several async sources — combine, fork-join, race, sequence
• Cancellable work, where a new input should abandon the request in flight
• Real-time data arriving over time — sockets, server-sent events, polling
• Retry and backoff policies expressed as part of the pipeline

Handled elsewhere:

• A single request with no stream semantics — async/await says it in one line, and wrapping it in an Observable buys nothing
• Local component state — a stream is the wrong shape for a value that is simply read and written
• Application state ownership — RxJS carries values between places; deciding where they live is a separate concern
• Framework binding — converting a stream into whatever the view layer renders is that layer's own API

---

<philosophy>

Everything is a stream: clicks, responses, timers, socket frames. An Observable is a description of work rather than the work itself, so a pipe() chain builds a pipeline and subscribe() is what starts it — which is why a pipe with no subscriber does nothing at all, and why the same Observable subscribed twice usually does its work twice.

That laziness is also the source of the most common surprise: anything eager placed inside the chain (a fetch call evaluated at construction time rather than inside defer) escapes the model and runs once, shared, whether anyone subscribed or not.

</philosophy>

---

<patterns>

Core patterns

Pattern 1: Creation, and keeping it lazy

Match the creation function to the source, and wrap anything eager in defer so each subscriber gets its own execution.

import {
  of,
  from,
  fromEvent,
  interval,
  timer,
  defer,
  EMPTY,
  throwError,
} from "rxjs";

of(1, 2, 3); // fixed values, then completes
from([1, 2, 3]); // any iterable, then completes
fromEvent(element, "click"); // never completes
interval(1000); // never completes
timer(2000); // one value, then completes

defer(() => from(fetch("/api/users"))); // fresh request per subscriber
from(fetch("/api/users")); // ❌ fires at construction, shared by all

Full code: examples/core.md

---

Pattern 2: Operators in a pipe, never a nested subscribe

const adults$ = users$.pipe(
  filter((user) => user.age >= 18),
  map((user) => user.name),
  distinctUntilChanged(),
  tap((name) => console.log(name)), // side effects live in tap
);

A .subscribe() inside a .subscribe() creates a subscription nothing holds a reference to, so nothing can end it — the flattening operators exist to replace exactly this shape.

Full code: examples/core.md

---

Pattern 3: Error recovery with catchError

const results$ = query$.pipe(
  switchMap((query) =>
    defer(() => from(search(query))).pipe(
      retry({ count: 3, delay: (_, n) => timer(1000 * 2 ** (n - 1)) }),
      catchError(() => of({ results: [], failed: true })), // inner — outer survives
    ),
  ),
);

catchError must return an Observable, and where it sits decides what survives: inside the inner pipe the outer stream continues, outside it the whole chain ends on the first failure.

Full code: examples/core.md

---

Pattern 4: Rate limiting

// Wait for a pause — user input
input$.pipe(debounceTime(300), distinctUntilChanged());

// Cap the rate — continuous events
fromEvent(window, "scroll").pipe(throttleTime(200));

debounceTime emits the last value after silence; throttleTime emits the first value of each window and drops the rest. distinctUntilChanged after a debounce is what stops a type-then-undo from re-issuing the same request.

Full code: examples/higher-order.md, inside the search pipeline. Operator comparison: reference.md

---

Pattern 5: Flattening operators

The decision is what should happen when a new outer value arrives while the inner Observable is still running.

Cancel the previous inner and switch    → switchMap    (search, navigation)
Run the new one alongside               → mergeMap     (uploads, fire-and-forget)
Queue it until the previous finishes    → concatMap    (ordered saves, message send)
Ignore it until the current one ends    → exhaustMap   (submit, login, refresh)

query$.pipe(switchMap((q) => search(q)));
files$.pipe(mergeMap((file) => upload(file), 3)); // 3 = concurrency limit
saves$.pipe(concatMap((data) => save(data)));
clicks$.pipe(exhaustMap(() => submit()));

Full code: examples/higher-order.md

---

Pattern 6: Subjects

A Subject is both Observable and Observer, which makes it the way to multicast — and the way to push values in from imperative code.

new Subject<T>(); // no initial value, no replay
new BehaviorSubject<T>(initial); // always has a current value; .getValue()
new ReplaySubject<T>(n, windowMs); // replays the last n (optionally, recent ones)
new AsyncSubject<T>(); // the final value only, and only on complete()

BehaviorSubject for state that always has a value, ReplaySubject for history a late subscriber needs, Subject for events with no past, AsyncSubject for one final result.

Full code: examples/subjects.md

---

Pattern 7: Cleanup with takeUntil

class Controller {
  private readonly destroy$ = new Subject<void>();

  start(): void {
    source$
      .pipe(
        switchMap((id) => this.load(id)),
        takeUntil(this.destroy$), // last — ends the inner subscription too
      )
      .subscribe((data) => this.render(data));
  }

  stop(): void {
    this.destroy$.next();
    this.destroy$.complete();
  }
}

One destroy$ ends every subscription that pipes through it. Full code: examples/memory-leaks.md

</patterns>

---

<red_flags>

Red flags

Breaks at runtime:

• An unhandled error terminates the chain permanently. Every subsequent emission is dropped silently, and the stream cannot be revived — only re-subscribed.
• catchError returning a plain value rather than an Observable is a type error; of(fallback) is the fix.
• takeUntil placed before a higher-order operator ends the outer source and leaves the inner subscription running. The cleanup code reads as correct and the leak is silent.
• Nested .subscribe() calls produce subscriptions with no reference, multiplying with each outer emission.
• firstValueFrom rejects when the Observable completes without emitting — the deprecated .toPromise() resolved undefined instead, so a migration changes the failure mode.

Surprising behaviour:

• from(promise) runs the promise at construction, not at subscription, and every subscriber gets the same settled result. defer(() => from(promise)) restores laziness.
• shareReplay(1) defaults refCount to false, so the source stays subscribed after the last subscriber leaves. shareReplay({ bufferSize: 1, refCount: true }) is almost always what was meant.
• combineLatest emits nothing until every source has emitted at least once — one silent source stalls the whole combination.
• forkJoin emits only when every source completes, so a Subject or an interval in the list means it never emits.
• retry re-subscribes to the source from the top, so a non-idempotent source runs its side effects again.
• interval(0) is still asynchronous — it schedules, it does not emit synchronously.
• BehaviorSubject.getValue() reads the current value without subscribing, which quietly turns reactive code into a poll.
• mergeMap where switchMap was meant produces a race: responses land out of order and the slowest one wins.

</red_flags>