# Web Utilities Rxjs
> Reactive programming with RxJS - Observables, operators, Subjects, error handling, and memory leak prevention
- Skill: `agents-inc/web-utilities-rxjs` (Agent Skill, multi-file: 8 files)
- Install (CLI): `npx skillmds@latest add agents-inc/web-utilities-rxjs`
- Raw SKILL.md: https://api.skillmd.com/api/skills/agents-inc/web-utilities-rxjs/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- Author: agents-inc (https://skillmd.com/u/agents-inc)
- Updated: 2026-09-10
- Page: https://skillmd.com/skills/agents-inc/web-utilities-rxjs
---
# 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](examples/core.md) — creation, lazy execution with `defer`, transformation, error recovery, Promise interop
- [examples/higher-order.md](examples/higher-order.md) — `switchMap`, `mergeMap`, `concatMap`, `exhaustMap` worked through
- [examples/subjects.md](examples/subjects.md) — the four Subject types, multicasting, `shareReplay`
- [examples/combination.md](examples/combination.md) — `combineLatest`, `forkJoin`, `merge`, `concat`, `race`
- [examples/memory-leaks.md](examples/memory-leaks.md) — the `takeUntil` pattern, subscription collections, self-completing streams
- [reference.md](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](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](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](examples/core.md).
---
## 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.
---
**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
---
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.
---
## 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.
```typescript
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](examples/core.md)
---
### Pattern 2: Operators in a pipe, never a nested subscribe
```typescript
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](examples/core.md)
---
### Pattern 3: Error recovery with catchError
```typescript
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](examples/core.md)
---
### Pattern 4: Rate limiting
```typescript
// 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](examples/higher-order.md), inside the search pipeline.
Operator comparison: [reference.md](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)
```
```typescript
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](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.
```typescript
new Subject(); // no initial value, no replay
new BehaviorSubject(initial); // always has a current value; .getValue()
new ReplaySubject(n, windowMs); // replays the last n (optionally, recent ones)
new AsyncSubject(); // 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](examples/subjects.md)
---
### Pattern 7: Cleanup with takeUntil
```typescript
class Controller {
private readonly destroy$ = new Subject();
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](examples/memory-leaks.md)
---
## 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.