RxJS Observables & Operators

Core Concepts

• Observable — lazy stream of values over time; nothing happens until subscribed

• Observer — { next, error, complete } callbacks

• Subject — both Observable and Observer; multicast (share one stream with many subscribers)

• BehaviorSubject — Subject with current value; emits last value to new subscribers

• Operator — pure function that transforms one Observable into another (map, filter, switchMap...)

Creation & Subscription

import {
  Observable, Subject, BehaviorSubject, ReplaySubject,
  of, from, interval, timer, fromEvent, forkJoin, combineLatest, merge,
  EMPTY, NEVER,
} from 'rxjs';
import {
  map, filter, take, takeUntil, switchMap, mergeMap, concatMap, exhaustMap,
  debounceTime, throttleTime, distinctUntilChanged, catchError, retry,
  tap, shareReplay, startWith, withLatestFrom, combineLatestWith,
} from 'rxjs/operators';

// Creating observables
of(1, 2, 3)                          // emits 1, 2, 3 then completes
from([1, 2, 3])                      // from iterable or Promise
from(fetch('/api/users'))            // from Promise
interval(1000)                       // 0, 1, 2... every second (never completes)
timer(2000, 1000)                    // starts after 2s, then every 1s
fromEvent(document, 'click')        // DOM event stream
EMPTY                                // completes immediately
NEVER                                // never emits, never completes

// Custom observable
const obs$ = new Observable<number>(subscriber => {
  subscriber.next(1);
  subscriber.next(2);
  setTimeout(() => { subscriber.next(3); subscriber.complete(); }, 1000);
  return () => console.log('cleanup');  // teardown logic
});

// Subscribe
const sub = obs$.subscribe({
  next: val => console.log('Value:', val),
  error: err => console.error('Error:', err),
  complete: () => console.log('Done'),
});
// Or shorthand: obs$.subscribe(val => console.log(val));

// Always unsubscribe to prevent memory leaks
sub.unsubscribe();

Key Operators

// Transformation
obs$.pipe(map(x => x * 2))
obs$.pipe(filter(x => x > 0))
obs$.pipe(take(5))                    // take first 5, then complete
obs$.pipe(takeUntil(stop$))           // complete when stop$ emits

// FlatMap variants (handle inner observables)
// switchMap  — cancel previous inner, start new (search autocomplete)
// mergeMap   — all inner run concurrently (no cancel)
// concatMap  — queue; one at a time in order (sequential operations)
// exhaustMap — ignore new while current inner is active (submit button)

const search$ = fromEvent<InputEvent>(input, 'input').pipe(
  map(e => (e.target as HTMLInputElement).value),
  debounceTime(300),               // wait 300ms after last keystroke
  distinctUntilChanged(),          // ignore if value didn't change
  filter(q => q.length >= 2),
  switchMap(q => from(fetch(`/api/search?q=${q}`).then(r => r.json()))),
  catchError(err => { console.error(err); return EMPTY; })
);

// Combining streams
combineLatest([user$, permissions$]).pipe(
  map(([user, perms]) => ({ ...user, permissions: perms }))
)
forkJoin([obs1$, obs2$, obs3$])      // wait for all to complete, emit last values
merge(click$, keypress$)             // interleave events from both

// Side effects (don't use tap for business logic)
obs$.pipe(tap(val => console.log('debug:', val)))

// Error handling
obs$.pipe(
  retry(3),                         // retry up to 3 times on error
  catchError(err => of(defaultValue)) // recover with fallback
)

// Sharing (prevent re-subscription / multiple HTTP calls)
const users$ = from(api.getUsers()).pipe(shareReplay(1));

// Subjects
const subject = new Subject<string>();
subject.next('hello');              // push value
subject.subscribe(console.log);    // subscribe

const state$ = new BehaviorSubject({ count: 0 });
state$.next({ count: 1 });
state$.getValue();                  // get current value synchronously