Observables & Operators

RxJS is a library for reactive programming using Observables — lazy push-based streams of values over time. Nothing executes until you subscribe.

Creating Observables

import {
  Observable, of, from, interval, timer, fromEvent, EMPTY, NEVER,
} from 'rxjs';

// of — emit a fixed sequence of values then complete
of(1, 2, 3).subscribe(console.log); // 1, 2, 3

// from — from array, iterable, or Promise
from([10, 20, 30]).subscribe(console.log);      // 10, 20, 30
from(fetch('/api/users').then(r => r.json()))   // from Promise
  .subscribe(users => console.log(users));

// interval — emit 0, 1, 2... every N ms (never completes)
interval(1000).subscribe(n => console.log(n)); // 0, 1, 2...

// timer — emit once after delay, or repeatedly after initial delay
timer(2000).subscribe(() => console.log('2s later')); // once
timer(0, 1000).subscribe(n => console.log(n));        // 0, 1, 2...

// fromEvent — DOM or Node.js EventEmitter event stream
fromEvent(document, 'click').subscribe(e => console.log(e));
fromEvent(document, 'keydown')
  .subscribe((e: Event) => console.log((e as KeyboardEvent).key));

// Custom Observable — full control over emission
const custom$ = new Observable<number>(subscriber => {
  subscriber.next(1);
  subscriber.next(2);
  setTimeout(() => {
    subscriber.next(3);
    subscriber.complete();
  }, 1000);
  // Return teardown function (called on unsubscribe)
  return () => console.log('cleaned up');
});

// EMPTY — completes immediately without emitting
// NEVER  — never emits, never completes (use in tests/stubs)
EMPTY.subscribe({ complete: () => console.log('done') });

Transformation & Filtering Operators

import { of, from, interval } from 'rxjs';
import {
  map, filter, take, skip, takeLast, first, last,
  distinctUntilChanged, debounceTime, throttleTime,
  scan, reduce, tap,
} from 'rxjs/operators';

// map — transform each value
of(1, 2, 3).pipe(map(x => x * 2)).subscribe(console.log); // 2, 4, 6

// filter — drop values that don't pass predicate
of(1, 2, 3, 4, 5).pipe(filter(x => x % 2 === 0)).subscribe(console.log); // 2, 4

// take / skip / first / last
interval(500).pipe(take(3)).subscribe(console.log); // 0, 1, 2 then complete
of(1,2,3,4,5).pipe(skip(2)).subscribe(console.log); // 3, 4, 5
of(1,2,3).pipe(first()).subscribe(console.log);     // 1
of(1,2,3).pipe(last()).subscribe(console.log);      // 3

// distinctUntilChanged — skip consecutive duplicates
of(1,1,2,3,3,2).pipe(distinctUntilChanged()).subscribe(console.log); // 1,2,3,2

// debounceTime — emit only after silence for N ms (search input)
// throttleTime — emit once per N ms window (scroll handler)

// scan — like reduce but emits running total
of(1, 2, 3, 4).pipe(
  scan((acc, val) => acc + val, 0)
).subscribe(console.log); // 1, 3, 6, 10

// tap — side effects without modifying stream (logging, debugging)
of(1, 2, 3).pipe(
  tap(x => console.log('before:', x)),
  map(x => x * 10),
  tap(x => console.log('after:', x))
).subscribe();

Higher-Order Mapping Operators

import { fromEvent, from, of, interval, EMPTY } from 'rxjs';
import { switchMap, mergeMap, concatMap, exhaustMap, catchError } from 'rxjs/operators';

// All four flatten an Observable<Observable<T>> to Observable<T>
// The difference is how they handle overlapping inner observables:

// switchMap  — cancel previous inner, start new (best for search/autocomplete)
// mergeMap   — all inner run concurrently, no ordering guarantee
// concatMap  — queue: one inner at a time, in order (sequential HTTP requests)
// exhaustMap — ignore new outer values while inner is active (submit button)

const input = document.querySelector('input')!;
const searchResults$ = fromEvent(input, 'input').pipe(
  map((e: Event) => (e.target as HTMLInputElement).value),
  debounceTime(300),
  distinctUntilChanged(),
  filter(q => q.length >= 2),
  switchMap(query =>
    from(fetch(`/api/search?q=${query}`).then(r => r.json())).pipe(
      catchError(err => { console.error(err); return EMPTY; })
    )
  )
);

// mergeMap — parallel requests (order not guaranteed)
const ids = [1, 2, 3];
from(ids).pipe(
  mergeMap(id => from(fetch(`/api/item/${id}`).then(r => r.json())))
).subscribe(item => console.log(item));

// concatMap — sequential (waits for previous to complete)
from([1, 2, 3]).pipe(
  concatMap(n => of(n).pipe(delay(500)))
).subscribe(console.log); // 1 (after 0.5s), 2 (after 1s), 3 (after 1.5s)

// exhaustMap — ignore clicks while request is in flight
const submitBtn = document.querySelector('button')!;
fromEvent(submitBtn, 'click').pipe(
  exhaustMap(() => from(fetch('/api/submit').then(r => r.json())))
).subscribe(response => console.log('submitted:', response));

Combining Streams & Error Handling

import { combineLatest, forkJoin, merge, zip, of, throwError } from 'rxjs';
import { catchError, retry, retryWhen, delay, take, shareReplay, startWith } from 'rxjs/operators';

// combineLatest — emit array whenever any source emits (all must emit at least once)
const user$ = of({ name: 'Alice' });
const perms$ = of(['read', 'write']);
combineLatest([user$, perms$]).pipe(
  map(([user, perms]) => ({ ...user, permissions: perms }))
).subscribe(console.log);

// forkJoin — wait for ALL to complete, emit last value of each (like Promise.all)
forkJoin({
  users: from(fetch('/api/users').then(r => r.json())),
  config: from(fetch('/api/config').then(r => r.json())),
}).subscribe(({ users, config }) => console.log(users, config));

// merge — interleave emissions from multiple sources
const click$ = fromEvent(document, 'click');
const key$ = fromEvent(document, 'keydown');
merge(click$, key$).subscribe(e => console.log('interaction', e.type));

// Error handling
of(1, 2, 3).pipe(
  map(x => { if (x === 2) throw new Error('bad value'); return x; }),
  catchError(err => {
    console.error('caught:', err.message);
    return of(-1); // recover with fallback value
  })
).subscribe(console.log); // 1, -1

// retry — resubscribe up to N times on error
from(fetch('/api/data')).pipe(
  retry(3),
  catchError(err => of({ error: true, message: err.message }))
).subscribe(console.log);

// shareReplay — multicast + replay last N values to new subscribers
// Prevents multiple HTTP calls when multiple components subscribe
const users$ = from(fetch('/api/users').then(r => r.json())).pipe(
  shareReplay(1) // cache the last emission
);
users$.subscribe(u => console.log('component A:', u));
users$.subscribe(u => console.log('component B:', u)); // no second HTTP call