RxJS & Observables

An observable is a description of something that will push values to you over time. It can push zero, one or many values, and then it either completes (finished normally) or errors (finished with a failure) — or it never finishes at all, like a stream of clicks. Angular uses observables for HTTP responses, router events, form value changes and anything else that "happens later, maybe more than once".

The single most useful mental model: an observable is a function, not a value. Creating one just writes down a recipe — "when someone subscribes, start a timer and call them back with numbers". Nothing runs until you call subscribe(), and every subscribe() runs the recipe again, from the top, just for that subscriber. Think of it as a newspaper subscription rather than a parcel: nothing is printed for you until you subscribe, you then get issue after issue, the paper may eventually close down (complete) or go bust (error), and you can cancel at any time (unsubscribe).

The contract every observable follows is next* (error | complete)? — any number of next values, then at most one of error or complete, and after that, silence. The function you return from the recipe is the teardown: RxJS calls it when the stream finishes or when the subscriber unsubscribes, which is where timers get cleared and requests get aborted.

import { Observable } from 'rxjs';

const start = Date.now();
const log = (...args) => console.log(`${String(Date.now() - start).padStart(3)}ms`, ...args);

const numbers$ = new Observable((subscriber) => {
  log('producer starts');
  subscriber.next(1);
  subscriber.next(2);
  const id = setTimeout(() => {
    subscriber.next(3);
    subscriber.complete();
    subscriber.next(4); // ignored: the stream already completed
  }, 100);
  return () => {
    log('teardown');
    clearTimeout(id);
  };
});

log('observable created');
numbers$.subscribe({
  next: (v) => log('next', v),
  complete: () => log('complete'),
});
log('subscribe() returned');

//   0ms observable created
//   0ms producer starts
//   0ms next 1
//   0ms next 2
//   0ms subscribe() returned
// 100ms next 3
// 100ms complete
// 100ms teardown

new Observable((s) => {
  s.next(1);
  s.complete();
  s.next(2);
  s.error(new Error('late'));
}).subscribe({
  next: (v) => console.log('next', v),
  error: (e) => console.log('error', e.message),
  complete: () => console.log('complete'),
});

Both represent "a result that arrives later", which is why they get compared in every interview. The differences all fall out of the mental model above: a promise is a value (a placeholder for one result that is already being produced), while an observable is a function (a recipe that runs per subscriber and can deliver many results).

import { Observable, interval, take, firstValueFrom, lastValueFrom } from 'rxjs';

const promise = new Promise((resolve) => {
  console.log('promise executor runs');
  resolve(1);
  resolve(2); // ignored: a promise settles once
});

const obs$ = new Observable((subscriber) => {
  console.log('observable producer runs');
  subscriber.next(1);
  subscriber.next(2);
  subscriber.complete();
});

console.log('--- nothing subscribed yet ---');
promise.then((v) => console.log('promise value', v));
obs$.subscribe((v) => console.log('observable value', v));
console.log('--- end of script ---');

setTimeout(async () => {
  console.log('first:', await firstValueFrom(interval(10).pipe(take(3))));
  console.log('last:', await lastValueFrom(interval(10).pipe(take(3))));
}, 20);

// promise executor runs
// --- nothing subscribed yet ---
// observable producer runs
// observable value 1
// observable value 2
// --- end of script ---
// promise value 1
// first: 0
// last: 2

const p = new Promise((resolve) => { console.log('P'); resolve(); });
const o = new Observable((s) => { console.log('O'); s.next(); });
console.log('end');

subscribe() is the moment an observable comes to life. You pass it an observer — an object with up to three callbacks, next, error and complete — or just a function, which is shorthand for { next: fn }. It returns a Subscription, your handle for stopping it: subscription.unsubscribe() runs the teardown and guarantees no more callbacks.

Who unsubscribes is the whole game in Angular. An observable that completes on its own (an HTTP request, of(...), take(3)) cleans itself up. One that never completes (an interval, a Subject, valueChanges, router events) keeps your callback — and everything it closes over, including your component — alive until someone unsubscribes. That's the classic RxJS memory leak.

import { Observable } from 'rxjs';

const ticks$ = new Observable((subscriber) => {
  let n = 0;
  console.log('timer started');
  const id = setInterval(() => subscriber.next(n++), 100);
  return () => {
    clearInterval(id);
    console.log('teardown: timer cleared');
  };
});

const sub = ticks$.subscribe({
  next: (n) => console.log('tick', n),
  error: (err) => console.log('error', err.message),
  complete: () => console.log('complete'), // never called: we unsubscribe instead
});

setTimeout(() => {
  sub.unsubscribe();
  console.log('closed?', sub.closed);
}, 350);

// timer started
// tick 0
// tick 1
// tick 2
// teardown: timer cleared
// closed? true

const sub = of(1, 2, 3).subscribe((v) => {
  console.log('got', v);
  sub.unsubscribe();
});
console.log('after subscribe');

My notes have this in capitals: don't subscribe multiple times — multiple subscriptions mean multiple executions. It follows directly from "an observable is a function": calling a function twice runs it twice. With HttpClient, every subscription is a separate HTTP request.

What makes it sneaky is that the extra subscriptions are often hidden. Two maps off the same source, two | async pipes on the same observable in a template, a subscribe() in a component plus an async pipe in its template — each is a separate trip to the server. pipe() doesn't share anything; it builds a new recipe that, when subscribed, subscribes to the source again.

import { Observable, map, shareReplay } from 'rxjs';

let executions = 0;
// Behaves like HttpClient.get(): nothing happens until subscribe, then one "request" per subscriber.
const getUser = () =>
  new Observable((subscriber) => {
    executions++;
    console.log(`request #${executions} sent`);
    const id = setTimeout(() => {
      subscriber.next({ id: 1, name: 'Ada', email: 'ada@example.com' });
      subscriber.complete();
    }, 100);
    return () => clearTimeout(id);
  });

const user$ = getUser();
const name$ = user$.pipe(map((u) => u.name));
const email$ = user$.pipe(map((u) => u.email));

name$.subscribe((n) => console.log('name:', n));
email$.subscribe((e) => console.log('email:', e));
// request #1 sent
// request #2 sent
// name: Ada
// email: ada@example.com
// → executions: 2

const shared$ = getUser().pipe(shareReplay({ bufferSize: 1, refCount: true }));
shared$.pipe(map((u) => u.name)).subscribe((n) => console.log('name:', n));
shared$.pipe(map((u) => u.email)).subscribe((e) => console.log('email:', e));
// request #1 sent
// name: Ada
// email: ada@example.com
// → executions: 1

// …later, after the response arrived:
shared$.subscribe((u) => console.log('late subscriber gets cached:', u.name));
// late subscriber gets cached: Ada
// → executions: still 1

// ❌ Two async pipes on the same cold observable = two subscriptions = two HTTP requests.
@Component({
  selector: 'app-user-twice',
  imports: [AsyncPipe],
  template: `
    <h2>{{ (user$ | async)?.name }}</h2>
    <p>{{ (user$ | async)?.email }}</p>
  `,
})
export class UserTwice {
  user$ = inject(UserApi).getUser(1);
}

// ✅ One async pipe, aliased with `as`.
@Component({
  selector: 'app-user-once',
  imports: [AsyncPipe],
  template: `
    @if (user$ | async; as user) {
      <h2>{{ user.name }}</h2>
      <p>{{ user.email }}</p>
    } @else {
      <p>Loading…</p>
    }
  `,
})
export class UserOnce {
  user$ = inject(UserApi).getUser(1);
}

let n = 0;
const req$ = new Observable((s) => { n++; s.next(n); s.complete(); });
const a$ = req$.pipe(map((x) => x * 10));
const b$ = a$.pipe(map((x) => x + 1));

a$.subscribe((v) => console.log('a', v));
b$.subscribe((v) => console.log('b', v));
console.log('executions', n);

A cold observable creates its producer per subscriber: HttpClient requests, timer, interval, of, from([...]). Each subscriber gets its own private execution, starting from the beginning. A hot observable has a producer that exists independently of subscribers: DOM events, a WebSocket that's already open, a Subject, router events. Subscribers tap into whatever is happening now; if you subscribe late, you miss what already went by.

Analogy: cold is a video on demand — everyone who presses play gets the film from the start, on their own stream. Hot is live TV — everyone watches the same broadcast, and tuning in late means you missed the opening. Multicasting operators turn cold into hot: share() and shareReplay() put a Subject in front of the source so many subscribers share one execution.

import { interval, take, share, timer } from 'rxjs';

// Cold: each subscriber gets its own interval, starting from 0.
const cold$ = interval(100).pipe(take(3));
cold$.subscribe((v) => log('A got', v));
timer(150).subscribe(() => cold$.subscribe((v) => log('B got', v)));
// 100ms A got 0
// 200ms A got 1
// 250ms B got 0
// 300ms A got 2
// 350ms B got 1
// 450ms B got 2

// Hot (shared): one interval, B joins mid-way.
const hot$ = interval(100).pipe(take(3), share());
hot$.subscribe((v) => log('A got', v));
timer(150).subscribe(() => hot$.subscribe((v) => log('B got', v)));
// 100ms A got 0
// 200ms A got 1
// 200ms B got 1
// 300ms A got 2
// 300ms B got 2

import { defer, interval, timer, map, tap, share, shareReplay } from 'rxjs';

let executions = 0;
const user$ = defer(() => {
  executions++;
  log(`request #${executions}`);
  return timer(100).pipe(map(() => 'Ada'));
});

const viaShare$ = user$.pipe(share());
viaShare$.subscribe((v) => log('share A:', v));
timer(200).subscribe(() => viaShare$.subscribe((v) => log('share B (late):', v)));

const viaReplay$ = user$.pipe(shareReplay({ bufferSize: 1, refCount: true }));
timer(400).subscribe(() => viaReplay$.subscribe((v) => log('replay A:', v)));
timer(600).subscribe(() => viaReplay$.subscribe((v) => log('replay B (late):', v)));

//   0ms request #1
// 100ms share A: Ada
// 200ms request #2
// 300ms share B (late): Ada
// 400ms request #3
// 500ms replay A: Ada
// 600ms replay B (late): Ada

import { interval, timer, tap, shareReplay } from 'rxjs';

const leaky$ = interval(100).pipe(tap((n) => log('leaky source tick', n)), shareReplay(1));
const tidy$ = interval(100).pipe(tap((n) => log('tidy source tick', n)), shareReplay({ bufferSize: 1, refCount: true }));

const a = leaky$.subscribe();
const b = tidy$.subscribe();
timer(250).subscribe(() => {
  a.unsubscribe();
  b.unsubscribe();
  log('both subscribers gone');
});

// 100ms leaky source tick 0
// 100ms tidy source tick 0
// 200ms leaky source tick 1
// 200ms tidy source tick 1
// 250ms both subscribers gone
// 300ms leaky source tick 2
// 400ms leaky source tick 3
// 500ms leaky source tick 4
// … forever

const t$ = interval(100).pipe(take(3), share());
t$.subscribe({ next: (v) => log('A', v), complete: () => log('A complete') });
timer(350).subscribe(() =>
  t$.subscribe({ next: (v) => log('B', v), complete: () => log('B complete') }),
);

100ms A 0
200ms A 1
300ms A 2
300ms A complete
450ms B 0
550ms B 1
650ms B 2
650ms B complete

An operator is just a function that takes an observable and returns a new observable. map((x) => x * 2) doesn't do anything to values by itself — it returns a function source$ => newObservable$, and the new observable, when subscribed, subscribes to source$ and forwards transformed values. pipe() chains these functions: source$.pipe(a, b, c) is exactly c(b(a(source$))), written left to right so it reads in the order data flows.

Two consequences matter every day. First, pipe() never changes the source — it returns a new observable, so obs$.pipe(map(...)) on its own line, with the result thrown away, does nothing. Second, pipe() is still lazy — it only builds a longer recipe; nothing runs until the final observable is subscribed. Picture a factory line that's been assembled but not switched on.

import { Observable, of, map, filter } from 'rxjs';

// A hand-written operator: a function from Observable to Observable.
function double() {
  return (source) =>
    new Observable((subscriber) => {
      console.log('double: subscribing to its source');
      const sub = source.subscribe({
        next: (v) => subscriber.next(v * 2),
        error: (e) => subscriber.error(e),
        complete: () => subscriber.complete(),
      });
      return () => sub.unsubscribe();
    });
}

const source$ = of(1, 2, 3);
const doubled$ = source$.pipe(double());
console.log('same object?', doubled$ === source$);
console.log('pipe() returned, nothing ran yet');
doubled$.subscribe((v) => console.log('doubled', v));
source$.subscribe((v) => console.log('original', v));

// pipe(a, b) is just b(a(source))
const viaPipe$ = source$.pipe(filter((n) => n % 2 === 1), map((n) => n * 10));
const byHand$ = map((n) => n * 10)(filter((n) => n % 2 === 1)(source$));
viaPipe$.subscribe((v) => console.log('viaPipe', v));
byHand$.subscribe((v) => console.log('byHand', v));

// same object? false
// pipe() returned, nothing ran yet
// double: subscribing to its source
// doubled 2
// doubled 4
// doubled 6
// original 1
// original 2
// original 3
// viaPipe 10
// viaPipe 30
// byHand 10
// byHand 30

save() {
  this.http.post('/api/todos', this.todo).pipe(tap(() => this.toast('Saved')));
  // ❌ Nothing is sent: the observable is created, piped, and thrown away.
}

const o$ = of(1, 2, 3);
o$.pipe(map((x) => x * 2));
o$.subscribe((v) => console.log(v));

My notes called these "confusing as hell", and I think the confusion has two sources. The first: they look like the array methods but run per value, over time — one value travels through the whole pipe before the next value even starts. The second, and the bigger one: when an observable emits an array (which is what http.get<User[]>() does), RxJS's map/filter work on the whole array as one value, and you need the array's map/filter inside them.

The roles, with a conveyor-belt picture — each value is a box riding past stations:

• map(fn) — a station that replaces each box with fn(box). One in, one out.
• filter(pred) — a station that lets a box through or throws it off. One in, one or zero out. The value itself is never changed.
• tap(fn) — a window where you look at each box (log it, store it) without touching it. Whatever fn returns is ignored; the original box continues.

import { of, map, filter, tap } from 'rxjs';

console.log('--- Array: each stage runs over the WHOLE array, then the next stage ---');
[1, 2, 3, 4]
  .filter((n) => { console.log('filter', n); return n % 2 === 0; })
  .map((n) => { console.log('map', n); return n * 10; })
  .forEach((n) => console.log('result', n));

console.log('--- RxJS: each VALUE runs through the whole pipe before the next value ---');
of(1, 2, 3, 4)
  .pipe(
    tap((n) => console.log('tap in', n)),
    filter((n) => n % 2 === 0),
    map((n) => n * 10),
    tap((n) => console.log('tap out', n)),
  )
  .subscribe((n) => console.log('result', n));

// --- Array: each stage runs over the WHOLE array, then the next stage ---
// filter 1
// filter 2
// filter 3
// filter 4
// map 2
// map 4
// result 20
// result 40
// --- RxJS: each VALUE runs through the whole pipe before the next value ---
// tap in 1
// tap in 2
// tap out 20
// result 20
// tap in 3
// tap in 4
// tap out 40
// result 40

interface User { name: string; active: boolean; }

// Like http.get<User[]>('/api/users'): ONE emission, and that emission is the whole array.
const users$: Observable<User[]> = of([
  { name: 'Ada', active: true },
  { name: 'Bob', active: false },
  { name: 'Cy', active: true },
]);

const wrong$ = users$.pipe(filter((user) => user.active)); // ❌ TS2339: Property 'active' does not exist on type 'User[]'.

const right$ = users$.pipe(map((users) => users.filter((user) => user.active)));
// emits once: [{ name: 'Ada', … }, { name: 'Cy', … }]

const nested$ = users$.pipe(map((users) => of(users.length)));
// type: Observable<Observable<number>> — map doesn't flatten

const tapped$ = users$.pipe(tap((users) => users.length));
// type: Observable<User[]> — tap's return value is ignored

of(1, 2, 3)
  .pipe(
    tap((x) => x * 100),
    filter((x) => x !== 2),
    map((x) => x + 1),
  )
  .subscribe((v) => console.log(v));

of and from are creation functions: they turn ordinary values into observables. Both emit synchronously (for synchronous inputs) and then complete. The difference is one level of unpacking:

• of(a, b, c) emits each argument exactly as it is. of([1, 2, 3]) emits one value: the array.
• from(x) takes one thing and emits what's inside it: each element of an array or iterable, each character of a string, the resolved value of a promise, or the values of another observable.

My notes also had "of vs off". There is no off in RxJS. The word comes from event-emitter APIs (emitter.off('click', handler) in Node, jQuery's .off()), where it removes a listener. In RxJS the equivalent of "off" is unsubscribing — and fromEvent() does the listener removal for you when you unsubscribe.

import { of, from } from 'rxjs';

const show = (label) => ({
  next: (v) => console.log(label, JSON.stringify(v)),
  complete: () => console.log(label, 'complete'),
});

of(1, 2, 3).subscribe(show('of(1, 2, 3)'));
of([1, 2, 3]).subscribe(show('of([1, 2, 3])'));
from([1, 2, 3]).subscribe(show('from([1, 2, 3])'));
from('hi').subscribe(show("from('hi')"));
of().subscribe(show('of()'));

console.log('before');
of('of value').subscribe((v) => console.log(v));
from(Promise.resolve('promise value')).subscribe((v) => console.log(v));
console.log('after');

// of(1, 2, 3) 1
// of(1, 2, 3) 2
// of(1, 2, 3) 3
// of(1, 2, 3) complete
// of([1, 2, 3]) [1,2,3]
// of([1, 2, 3]) complete
// from([1, 2, 3]) 1
// from([1, 2, 3]) 2
// from([1, 2, 3]) 3
// from([1, 2, 3]) complete
// from('hi') "h"
// from('hi') "i"
// from('hi') complete
// of() complete
// before
// of value
// after
// promise value

import { of, from, defer } from 'rxjs';

let calls = 0;
const roll = () => ++calls;
const eager$ = of(roll());
const lazy$ = defer(() => of(roll()));
console.log('calls after creating both:', calls);
eager$.subscribe((v) => console.log('eager', v));
eager$.subscribe((v) => console.log('eager', v));
lazy$.subscribe((v) => console.log('lazy', v));
lazy$.subscribe((v) => console.log('lazy', v));
// calls after creating both: 1
// eager 1
// eager 1
// lazy 2
// lazy 3

let fetches = 0;
const fakeFetch = () => new Promise((resolve) => {
  fetches++;
  console.log('fetch started');
  setTimeout(() => resolve('data'), 10);
});
const p$ = from(fakeFetch());
console.log('fetches before subscribe:', fetches);
p$.subscribe((v) => console.log('sub 1', v));
p$.subscribe((v) => console.log('sub 2', v));
// fetch started
// fetches before subscribe: 1
// sub 1 data
// sub 2 data
// (later) defer(() => fakeFetch()) subscribed twice → "fetch started" twice, total fetches: 3

import { fromEvent } from 'rxjs';
import { EventEmitter } from 'node:events';

const emitter = new EventEmitter();
const clicks$ = fromEvent(emitter, 'click');
const sub = clicks$.subscribe((e) => console.log('click', e));
console.log('listeners while subscribed:', emitter.listenerCount('click'));
emitter.emit('click', 'first');
sub.unsubscribe();
console.log('listeners after unsubscribe:', emitter.listenerCount('click'));
emitter.emit('click', 'second (nobody hears this)');

// listeners while subscribed: 1
// click first
// listeners after unsubscribe: 0

from([[1, 2], [3]]).subscribe((v) => console.log('A', JSON.stringify(v)));
of(...[1, 2]).subscribe((v) => console.log('B', v));
from(of(1, 2)).subscribe((v) => console.log('C', v));
of(of(1)).subscribe((v) => console.log('D', v instanceof Observable));

An error is a terminal notification: once an observable errors, it's finished — no more values, ever, and every operator between the error and the subscriber is unsubscribed. If nothing handles it, the subscriber's error callback gets it, or RxJS reports it as unhandled.

catchError(handler) intercepts an error and replaces the failed stream with another observable returned by handler. Return of(fallback) to substitute a value, EMPTY to swallow it and complete, or throwError(() => newError) to rethrow something friendlier. What catchError can't do is "resume" the original stream: the source that errored is gone. finalize(fn) is the observable version of finally: it runs once when the stream ends for any reason — complete, error, or unsubscribe.

import { of, map, catchError, finalize, throwError, EMPTY, timer } from 'rxjs';

const watch = (label) => ({
  next: (v) => console.log(label, 'next', v),
  error: (e) => console.log(label, 'error', e.message),
  complete: () => console.log(label, 'complete'),
});
const failOn2 = map((n) => { if (n === 2) throw new Error('bad 2'); return n; });

of(1, 2, 3).pipe(failOn2).subscribe(watch('A'));
// A next 1
// A error bad 2

of(1, 2, 3)
  .pipe(failOn2, catchError((err) => { console.log('caught', err.message); return of(-1); }))
  .subscribe(watch('B'));
// B next 1
// caught bad 2
// B next -1
// B complete

throwError(() => new Error('nope')).pipe(catchError(() => EMPTY)).subscribe(watch('C'));
// C complete

throwError(() => new Error('500 Internal Server Error'))
  .pipe(catchError((err) => throwError(() => new Error(`Could not load users (${err.message})`))))
  .subscribe(watch('D'));
// D error Could not load users (500 Internal Server Error)

of('x').pipe(finalize(() => console.log('E finalize'))).subscribe(watch('E'));
// E next x
// E complete
// E finalize
throwError(() => new Error('fail')).pipe(finalize(() => console.log('F finalize'))).subscribe(watch('F'));
// F error fail
// F finalize
timer(1000).pipe(finalize(() => console.log('G finalize'))).subscribe(watch('G')).unsubscribe();
// G finalize

of(1, 2, 3)
  .pipe(
    map((x) => { if (x === 2) throw new Error('two'); return x; }),
    catchError(() => of(99)),
    map((x) => x * 10),
  )
  .subscribe({ next: (v) => console.log(v), complete: () => console.log('complete') });

You rarely want subscribe() in a component, because then you own the unsubscribe. Angular gives you two consumers that manage the subscription for you. The async pipe subscribes when the template renders the expression, gives the template the latest value (or null before the first one), marks the view for check on each emission, and unsubscribes when the view is destroyed. toSignal(obs$) does the same job from the class: it subscribes immediately, exposes the latest value as a signal, and unsubscribes when the component (its injection context) is destroyed. Because the result is a signal, you can read it in computed(), in templates without a pipe, and as many times as you like — one subscription no matter how often it's read.

The other direction exists too: toObservable(signal) gives you an observable of a signal's values, so a signal (say, a search query) can feed RxJS operators like debounceTime and switchMap. My notes call the async pipe a "head twister"; chapter 5's The async pipe covers the pipe itself, and Signals don't need the async pipe why signals skip it.

@Component({
  selector: 'app-user-signal',
  template: `
    @if (user(); as user) {
      <h2>{{ user.name }}</h2>
      <p>{{ user.email }}</p>
    } @else {
      <p>Loading…</p>
    }
  `,
})
export class UserSignal {
  // Subscribes once, right now (we're in an injection context), and unsubscribes on destroy.
  user = toSignal(inject(UserApi).getUser(1), { initialValue: null });
}

it('toObservable emits only the latest value per effect run', async () => {
  const count = signal(0);
  const seen: number[] = [];
  const count$ = TestBed.runInInjectionContext(() => toObservable(count));
  count$.subscribe((n) => seen.push(n));
  // seen: []          right after subscribe
  TestBed.tick();
  // seen: [0]
  count.set(1);
  count.set(2);
  count.set(3);
  // seen: [0]         before the next tick
  TestBed.tick();
  // seen: [0, 3]
  expect(seen).toEqual([0, 3]);
});

export const ticks = new Subject<number>();

@Component({ selector: 'app-ticker', template: `{{ last() }}` })
export class Ticker {
  last = signal(0);

  constructor() {
    ticks.pipe(takeUntilDestroyed()).subscribe((n) => this.last.set(n));
  }
}

@Component({ selector: 'app-late-ticker-fixed', template: `` })
export class LateTickerFixed implements OnInit {
  private destroyRef = inject(DestroyRef);

  ngOnInit() {
    ticks.pipe(takeUntilDestroyed(this.destroyRef)).subscribe();
  }
}

My notes open this part with "Chalo bro, RxJS part 2 — the famous four flattening operators." Here's the problem all four solve. Often each value from one stream should start another stream: each search term starts an HTTP request, each click starts a save. With plain map you'd get an observable of observables (map, filter and tap). A flattening operator (also called a higher-order mapping operator) subscribes to each inner observable for you and merges its values into one output stream. switchMap, mergeMap, concatMap and exhaustMap all do that; they differ only in what happens when a new outer value arrives while an inner one is still running.

switchMap cancels the previous inner observable and switches to the new one. Only the latest request matters. Think of a TV remote: pressing a new channel doesn't queue it after the current programme, it switches immediately and you stop seeing the old one. That's exactly right for searching, autocomplete and "load the details of whatever's selected now".

All four operators in this chapter run on the same scenario, so you can compare them line by line: the user types a at 0ms, ab at 50ms and abc at 120ms, and each search request takes 100ms.

import { merge, timer, map, tap } from 'rxjs';

const start = Date.now();
const log = (...args) => console.log(`${String(Date.now() - start).padStart(3)}ms`, ...args);

// One value at a given time.
const at = (ms, value) => timer(ms).pipe(map(() => value));

// Keystrokes: "a" at 0ms, "ab" at 50ms, "abc" at 120ms.
const searchBox$ = merge(at(0, 'a'), at(50, 'ab'), at(120, 'abc')).pipe(
  tap((q) => log(`typed "${q}"`)),
);

// A fake HTTP call: subscribing starts it, it answers after 100ms, unsubscribing cancels it.
const fakeSearch = (q) =>
  timer(100).pipe(
    map(() => `results for "${q}"`),
    tap({
      subscribe: () => log(`  request "${q}" started`),
      unsubscribe: () => log(`  request "${q}" cancelled`),
    }),
  );

const print = { next: (r) => log('→', r), complete: () => log('complete') };

searchBox$.pipe(switchMap((q) => fakeSearch(q))).subscribe(print);

//   0ms typed "a"
//   0ms   request "a" started
//  50ms typed "ab"
//  50ms   request "a" cancelled
//  50ms   request "ab" started
// 120ms typed "abc"
// 120ms   request "ab" cancelled
// 120ms   request "abc" started
// 220ms → results for "abc"
// 220ms complete

// A promise-based "fetch": starts immediately and cannot be stopped.
const promiseSearch = (q) =>
  new Promise((resolve) => {
    log(`  server starts "${q}"`);
    setTimeout(() => {
      log(`  server finished "${q}"`);
      resolve(`results for "${q}"`);
    }, 100);
  });

searchBox$.pipe(switchMap((q) => promiseSearch(q))).subscribe(print);

//   0ms typed "a"
//   0ms   server starts "a"
//  50ms typed "ab"
//  50ms   server starts "ab"
// 100ms   server finished "a"
// 120ms typed "abc"
// 120ms   server starts "abc"
// 150ms   server finished "ab"
// 220ms   server finished "abc"
// 220ms → results for "abc"
// 220ms complete

searchBox$
  .pipe(
    switchMap((q) => {
      fakeSearch(q).subscribe((r) => log('  nested subscribe still got', r)); // ❌ not managed by switchMap
      return of(q);
    }),
  )
  .subscribe((q) => log('switchMap emitted', q));

//   0ms typed "a"
//   0ms   request "a" started
//   0ms switchMap emitted a
//  50ms typed "ab"
//  50ms   request "ab" started
//  50ms switchMap emitted ab
// 100ms   nested subscribe still got results for "a"
// 120ms typed "abc"
// 120ms   request "abc" started
// 120ms switchMap emitted abc
// 150ms   nested subscribe still got results for "ab"
// 220ms   nested subscribe still got results for "abc"

const failing = (q) =>
  q === 'ab'
    ? timer(30).pipe(map(() => { throw new Error(`500 for "${q}"`); }))
    : fakeSearch(q);

// catchError OUTSIDE switchMap
searchBox$
  .pipe(switchMap(failing), catchError(() => { log('outer catchError'); return EMPTY; }))
  .subscribe(print);
//   0ms typed "a"
//   0ms   request "a" started
//  50ms typed "ab"
//  50ms   request "a" cancelled
//  80ms outer catchError
//  80ms complete

// catchError INSIDE switchMap, on the inner request
searchBox$
  .pipe(
    switchMap((q) =>
      failing(q).pipe(catchError((e) => { log('inner catchError:', e.message); return of(`no results for "${q}"`); })),
    ),
  )
  .subscribe(print);
//   0ms typed "a"
//   0ms   request "a" started
//  50ms typed "ab"
//  50ms   request "a" cancelled
//  80ms inner catchError: 500 for "ab"
//  80ms → no results for "ab"
// 120ms typed "abc"
// 120ms   request "abc" started
// 220ms → results for "abc"
// 220ms complete

box.query.set('rx');
TestBed.tick();
vi.advanceTimersByTime(300);
const first = http.expectOne((r) => r.url === '/api/search'); // /api/search?q=rx

box.query.set('rxjs');
TestBed.tick();
vi.advanceTimersByTime(300);
const second = http.expectOne((r) => r.url === '/api/search'); // /api/search?q=rxjs

console.log(first.cancelled, second.cancelled); // true false

merge(at(0, 'a'), at(50, 'b'))
  .pipe(switchMap((k) => interval(20).pipe(take(3), map((i) => `${k}${i}`))))
  .subscribe({ next: (v) => log(v), complete: () => log('complete') });

 20ms a0
 40ms a1
 70ms b0
 90ms b1
110ms b2
110ms complete

A typeahead shouldn't fire a request on every keystroke. debounceTime(ms) waits until the source has been quiet for ms, then emits the latest value: each new value restarts the countdown, so a burst of typing produces one emission, ms after the last key. distinctUntilChanged() drops a value if it's equal (===) to the previous emitted one — so if the debounced text comes out the same as last time (type a letter, delete it, pause), no new request is made.

Analogy for debounceTime: a lift door. Every person who walks in resets the door timer; the lift leaves only once nobody has come in for a few seconds. distinctUntilChanged is a doorman who says "you're the same person who just left, no need to go again".

import { NEVER, merge, timer, map, tap, debounceTime, distinctUntilChanged } from 'rxjs';

// r, rx, rxj typed quickly; a pause; then "rxjs", and a backspace back to "rxj".
const input$ = merge(at(0, 'r'), at(80, 'rx'), at(160, 'rxj'), at(600, 'rxjs'), at(700, 'rxj'), NEVER).pipe(
  tap((v) => log(`input "${v}"`)),
);

input$
  .pipe(
    debounceTime(300),
    tap((v) => log(`debounced "${v}"`)),
    distinctUntilChanged(),
  )
  .subscribe((v) => log(`distinct "${v}"`));

//    0ms input "r"
//   80ms input "rx"
//  160ms input "rxj"
//  460ms debounced "rxj"
//  460ms distinct "rxj"
//  600ms input "rxjs"
//  700ms input "rxj"
// 1000ms debounced "rxj"

merge(at(0, { q: 'rx' }), at(10, { q: 'rx' }), at(20, { q: 'rxjs' }))
  .pipe(distinctUntilChanged())
  .subscribe((v) => log('by reference:', v.q));
//  0ms by reference: rx
// 10ms by reference: rx
// 20ms by reference: rxjs

merge(at(100, { q: 'rx' }), at(110, { q: 'rx' }), at(120, { q: 'rxjs' }))
  .pipe(distinctUntilChanged((a, b) => a.q === b.q))
  .subscribe((v) => log('by comparator:', v.q));
// 100ms by comparator: rx
// 120ms by comparator: rxjs

@Component({
  selector: 'app-search-box',
  template: `
    <input [value]="query()" (input)="query.set($any($event.target).value)" />
    <ul>
      @for (name of results(); track name) {
        <li>{{ name }}</li>
      }
    </ul>
  `,
})
export class SearchBox {
  private api = inject(UserApi);
  query = signal('');

  results = toSignal(
    toObservable(this.query).pipe(
      map((text) => text.trim()),
      debounceTime(300),
      distinctUntilChanged(),
      filter((text) => text.length >= 2),
      switchMap((text) => this.api.search(text).pipe(catchError(() => of([])))),
    ),
    { initialValue: [] },
  );
}

merge(at(0, 'a'), at(100, 'b'), at(250, 'c'), at(400, 'd'), NEVER)
  .pipe(debounceTime(200))
  .subscribe((v) => log(v));

mergeMap runs every inner observable at the same time and merges their results as they arrive. Nothing is cancelled, nothing waits. (flatMap is its old, deprecated alias.) Think of a team of waiters taking orders: each new table gets served in parallel, and dishes come out in whatever order the kitchen finishes them.

That makes it right when every inner result matters and order doesn't: deleting several items, uploading several files, fetching details for each id in a list, handling independent messages. It's wrong for a typeahead (stale results can arrive last) and for writes to the same resource (they can be applied out of order).

searchBox$.pipe(mergeMap((q) => fakeSearch(q))).subscribe(print);

//   0ms typed "a"
//   0ms   request "a" started
//  50ms typed "ab"
//  50ms   request "ab" started
// 100ms → results for "a"
// 120ms typed "abc"
// 120ms   request "abc" started
// 150ms → results for "ab"
// 220ms → results for "abc"
// 220ms complete

const job = (name, ms) => timer(ms).pipe(map(() => name), tap({ subscribe: () => log(`  start ${name} (${ms}ms)`) }));
const durations = { a: 150, ab: 30 };

merge(at(0, 'a'), at(50, 'ab'))
  .pipe(mergeMap((q) => job(q, durations[q])))
  .subscribe((r) => log('→', r));

//   0ms   start a (150ms)
//  50ms   start ab (30ms)
//  80ms → ab
// 150ms → a

merge(at(0, 'u1'), at(0, 'u2'), at(0, 'u3'), at(0, 'u4'))
  .pipe(mergeMap((id) => job(id, 100), 2))
  .subscribe((r) => log('→ uploaded', r));

//   0ms   start u1 (100ms)
//   0ms   start u2 (100ms)
// 100ms → uploaded u1
// 100ms   start u3 (100ms)
// 100ms → uploaded u2
// 100ms   start u4 (100ms)
// 200ms → uploaded u3
// 200ms → uploaded u4

merge(at(0, 'a'), at(50, 'b'))
  .pipe(mergeMap((k) => interval(20).pipe(take(3), map((i) => `${k}${i}`))))
  .subscribe({ next: (v) => log(v), complete: () => log('complete') });

 20ms a0
 40ms a1
 60ms a2
 70ms b0
 90ms b1
110ms b2
110ms complete

My notes: "concatMap (order matters!!!)". concatMap runs inner observables one at a time, in arrival order. A new outer value waits in a queue until the current inner completes; nothing is cancelled and nothing overlaps. It's a single checkout till: everyone gets served, strictly in the order they joined the queue.

It's the safe default for writes where order matters: autosave, sequential PUTs to the same resource, steps that depend on the previous one finishing, an animation sequence.

searchBox$.pipe(concatMap((q) => fakeSearch(q))).subscribe(print);

//   0ms typed "a"
//   0ms   request "a" started
//  50ms typed "ab"
// 100ms → results for "a"
// 100ms   request "ab" started
// 120ms typed "abc"
// 200ms → results for "ab"
// 200ms   request "abc" started
// 300ms → results for "abc"
// 300ms complete

let serverState = 'v0';
const durations = { v1: 100, v2: 30 };
const save = (v) =>
  timer(durations[v]).pipe(
    tap({ subscribe: () => log(`  PUT ${v} sent`) }),
    tap(() => { serverState = v; log(`  PUT ${v} done → server has ${v}`); }),
  );

merge(at(0, 'v1'), at(10, 'v2'))
  .pipe(mergeMap(save)) // then try concatMap(save)
  .subscribe({ complete: () => log(`final server state: ${serverState}`) });

// with mergeMap:
//   0ms   PUT v1 sent
//  10ms   PUT v2 sent
//  40ms   PUT v2 done → server has v2
// 100ms   PUT v1 done → server has v1
// 100ms final server state: v1

// with concatMap:
//   0ms   PUT v1 sent
// 100ms   PUT v1 done → server has v1
// 100ms   PUT v2 sent
// 130ms   PUT v2 done → server has v2
// 130ms final server state: v2

merge(at(0, 'a'), at(50, 'b'))
  .pipe(concatMap((k) => interval(20).pipe(take(3), map((i) => `${k}${i}`))))
  .subscribe({ next: (v) => log(v), complete: () => log('complete') });

 20ms a0
 40ms a1
 60ms a2
 80ms b0
100ms b1
120ms b2
120ms complete

My notes: "exhaustMap (wait until I am done)". exhaustMap ignores new outer values while an inner observable is running. No cancelling, no queue: while it's busy, incoming values are simply dropped. Once the inner completes, the next value to arrive starts a new one. Picture a lift that's already moving: pressing the call button again does nothing until it's back.

It's the right tool when a repeated trigger should not start the same work again: a login or submit button that gets double-clicked, a "refresh" button, a pull-to-refresh gesture.

searchBox$.pipe(exhaustMap((q) => fakeSearch(q))).subscribe(print);

//   0ms typed "a"
//   0ms   request "a" started
//  50ms typed "ab"
// 100ms → results for "a"
// 120ms typed "abc"
// 120ms   request "abc" started
// 220ms → results for "abc"
// 220ms complete

merge(at(0, 'click'), at(30, 'click'), at(60, 'click'), at(250, 'click'), NEVER)
  .pipe(
    tap(() => log('click')),
    exhaustMap(() => job('login request', 200)),
  )
  .subscribe(() => log('→ logged in'));

//   0ms click
//   0ms   start login request (200ms)
//  30ms click
//  60ms click
// 200ms → logged in
// 250ms click
// 250ms   start login request (200ms)
// 450ms → logged in

merge(at(0, 'a'), at(50, 'b'))
  .pipe(exhaustMap((k) => interval(20).pipe(take(3), map((i) => `${k}${i}`))))
  .subscribe({ next: (v) => log(v), complete: () => log('complete') });

20ms a0
40ms a1
60ms a2
60ms complete

All four take the same arguments and subscribe to the inner observable for you. The only question is: a new value arrives while the previous inner is still running — what should happen to the old one and the new one?