web-utilities-rxjs

RxJS Reactive Programming Patterns

Quick Guide: Use RxJS for complex async flows involving event streams, composed async operations, and reactive data pipelines. Create Observables with of, from, fromEvent, or defer. Transform with pipeable operators (map, filter, switchMap). Always unsubscribe to prevent memory leaks -- use the takeUntil pattern with a destroy Subject, or store subscriptions and call unsubscribe() in cleanup. Suffix observable variables with $. Choose the right flattening operator: switchMap to cancel previous, mergeMap for parallel, concatMap for sequential, exhaustMap to ignore while busy.

<critical_requirements>

CRITICAL: Before Using This Skill

All code must follow project conventions in CLAUDE.md (kebab-case, named exports, import ordering, import type, named constants)

(You MUST unsubscribe from every long-lived Observable -- use takeUntil, take, first, or explicit unsubscribe() to prevent memory leaks)

(You MUST choose the correct flattening operator -- switchMap for cancellation, mergeMap for parallel, concatMap for sequential, exhaustMap to ignore while busy)

(You MUST place takeUntil LAST in the pipe chain -- placing it before higher-order operators like switchMap leaves inner subscriptions alive)

(You MUST handle errors with catchError inside the pipe -- an unhandled error terminates the entire Observable chain permanently)

</critical_requirements>

Auto-detection: RxJS, rxjs, Observable, Subject, BehaviorSubject, ReplaySubject, AsyncSubject, subscribe, pipe, switchMap, mergeMap, concatMap, exhaustMap, combineLatest, forkJoin, fromEvent, of, from, interval, timer, defer, catchError, retry, debounceTime, throttleTime, distinctUntilChanged, takeUntil, map, filter, tap, shareReplay

When to use:

• Complex async flows composing multiple HTTP requests, WebSockets, or event streams
• Event stream processing with debounce, throttle, or buffer logic
• Real-time data pipelines (stock tickers, chat, live updates)
• Cancellable HTTP requests (type-ahead search, route changes)
• Coordinating multiple async sources (race, combine, fork-join)

When NOT to use:

• Simple one-shot HTTP requests (use fetch or your data-fetching solution)
• Simple component state (use useState, ref, or signals)
• State management (use a dedicated state management solution)
• Single async operation with no stream composition (use async/await)

Key patterns covered:

• Observable creation (of, from, fromEvent, interval, timer, defer)
• Pipeable operators (transformation, filtering, combination)
• Higher-order mapping (switchMap, mergeMap, concatMap, exhaustMap)
• Subjects (Subject, BehaviorSubject, ReplaySubject, AsyncSubject)
• Error handling (catchError, retry with config, retryWhen deprecated)
• Memory leak prevention (takeUntil pattern, subscription management)
• Combination operators (combineLatest, forkJoin, merge, concat)

Detailed Resources:

• [examples/core.md](examples/core.md) - Observable creation, subscription, basic operators, error handling
• [examples/higher-order.md](examples/higher-order.md) - switchMap, mergeMap, concatMap, exhaustMap patterns
• [examples/subjects.md](examples/subjects.md) - Subject types, multicasting, state sharing
• [examples/combination.md](examples/combination.md) - combineLatest, forkJoin, merge, concat, race
• [examples/memory-leaks.md](examples/memory-leaks.md) - takeUntil pattern, subscription cleanup strategies
• [reference.md](reference.md) - Operator decision tree, flattening operator cheat sheet, gotchas

<philosophy>

Philosophy

RxJS implements the Observable pattern for composing asynchronous and event-based programs using observable sequences. The core idea: treat everything as a stream -- clicks, HTTP responses, timers, WebSocket messages -- and compose them declaratively with operators.

Key principle: Observables are lazy. Nothing happens until you subscribe(). Operators in a pipe() build a processing pipeline that transforms values as they flow through.

import { fromEvent, map, filter, debounceTime } from "rxjs";

const DEBOUNCE_MS = 300;
const MIN_QUERY_LENGTH = 3;

// Declarative stream: input events -> debounced search queries
const search$ = fromEvent<InputEvent>(searchInput, "input").pipe(
  map((event) => (event.target as HTMLInputElement).value),
  debounceTime(DEBOUNCE_MS),
  filter((query) => query.length >= MIN_QUERY_LENGTH),
);

Naming Convention

Suffix observable variables with $ to distinguish them from plain values:

const users$ = userService.getUsers(); // Observable
const users = []; // plain array

Current Version

RxJS 7.8.x is the current stable release. v7 introduced smaller bundle sizes (~50% smaller via tree-shaking), pipeable-only operators, and improved TypeScript types. v8 is in development.

When to Use RxJS

• Event streams that need composition (debounce + filter + switchMap)
• Multiple concurrent async operations needing coordination
• Real-time data that arrives over time (WebSockets, SSE, polling)
• Complex cancellation logic (cancel previous request on new input)

When NOT to Use RxJS

• Simple fetch calls -- async/await or your data-fetching solution is simpler
• One-off async operations without stream semantics
• State management -- use a dedicated state solution
• When the team lacks RxJS experience and the problem is simple

</philosophy>

<patterns>

Core Patterns

Pattern 1: Observable Creation

Use the right creation function for the data source.

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

// ✅ Good Example - Matching creation function to data source

// Static values
const static$ = of(1, 2, 3); // emits 1, 2, 3, then completes

// From iterable/Promise/array
const fromArray$ = from([1, 2, 3]); // emits 1, 2, 3, then completes
const fromPromise$ = from(fetch("/api/users")); // emits response, completes

// DOM events (infinite stream -- must unsubscribe)
const clicks$ = fromEvent(document, "click");

// Timed emissions
const POLL_INTERVAL_MS = 1000;
const interval$ = interval(POLL_INTERVAL_MS); // emits 0, 1, 2... every second

const INITIAL_DELAY_MS = 2000;
const timer$ = timer(INITIAL_DELAY_MS); // emits 0 after 2s, then completes

// Deferred creation -- new Observable per subscriber
const deferred$ = defer(() => from(fetch("/api/data")));

// Sentinel observables
const empty$ = EMPTY; // completes immediately, emits nothing
const error$ = throwError(() => new Error("Something failed"));

Why good: each creation function matches the data source semantics, defer ensures fresh execution per subscriber, named constants for durations

// ❌ Bad Example - Wrapping a promise without defer
const bad$ = from(fetch("/api/users")); // fetch fires immediately, shared across subscribers!

Why bad: without defer, the fetch executes once when the Observable is created, not when subscribed -- all subscribers share the same stale response

Pattern 2: Pipeable Operators

All operators are used inside .pipe(). Import from "rxjs/operators" or "rxjs" (v7.2+).

import { map, filter, tap, take, skip, distinctUntilChanged } from "rxjs";

const MIN_AGE = 18;

// ✅ Good Example - Operator pipeline
const adults$ = users$.pipe(
  filter((user) => user.age >= MIN_AGE),
  map((user) => user.name),
  distinctUntilChanged(), // skip consecutive duplicates
  tap((name) => console.log("Processing:", name)), // side effects (logging)
);

Why good: clear transformation pipeline, each operator has a single purpose, tap for side effects keeps the pipeline pure otherwise

// ❌ Bad Example - Nesting subscribes instead of using operators
users$.subscribe((us