Angular RxJS & Observables

RxJS (Reactive Extensions for JavaScript) is a library for reactive programming using Observables. Angular extensively uses RxJS for handling asynchronous operations, HTTP requests, and event handling.

Observable Basics

import { Observable, Observer, of, from, interval } from 'rxjs';

// Create observable from scratch
const customObservable$ = new Observable<number>((observer: Observer<number>) => {
  let count = 0;
  const intervalId = setInterval(() => {
    observer.next(count++);
    
    if (count === 5) {
      observer.complete();
      clearInterval(intervalId);
    }
  }, 1000);
  
  // Cleanup function
  return () => {
    clearInterval(intervalId);
  };
});

// Subscribe to observable
const subscription = customObservable$.subscribe({
  next: (value) => console.log('Value:', value),
  error: (err) => console.error('Error:', err),
  complete: () => console.log('Complete')
});

// Unsubscribe
subscription.unsubscribe();

// Create observables from values
const fromValue$ = of(1, 2, 3, 4, 5);
const fromArray$ = from([1, 2, 3, 4, 5]);
const fromPromise$ = from(fetch('/api/data'));
const timer$ = interval(1000); // Emits every second

Common RxJS Operators

Transformation Operators

import { map, pluck, scan, reduce } from 'rxjs/operators';
import { of } from 'rxjs';

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

// pluck - extract property
of(
  { name: 'John', age: 30 },
  { name: 'Jane', age: 25 }
)
  .pipe(
    pluck('name')
  )
  .subscribe(console.log); // 'John', 'Jane'

// scan - accumulate values (like reduce but emits intermediate values)
of(1, 2, 3, 4, 5)
  .pipe(
    scan((acc, curr) => acc + curr, 0)
  )
  .subscribe(console.log); // 1, 3, 6, 10, 15

// reduce - accumulate and emit final value
of(1, 2, 3, 4, 5)
  .pipe(
    reduce((acc, curr) => acc + curr, 0)
  )
  .subscribe(console.log); // 15

Filtering Operators

import { filter, take, takeUntil, takeWhile, skip, distinct, distinctUntilChanged } from 'rxjs/operators';
import { interval, Subject } from 'rxjs';

// filter - emit values that pass condition
of(1, 2, 3, 4, 5)
  .pipe(
    filter(x => x % 2 === 0)
  )
  .subscribe(console.log); // 2, 4

// take - emit first N values
interval(1000)
  .pipe(
    take(3)
  )
  .subscribe(console.log); // 0, 1, 2

// takeUntil - emit until notifier emits
const stop$ = new Subject();
interval(1000)
  .pipe(
    takeUntil(stop$)
  )
  .subscribe(console.log);

setTimeout(() => stop$.next(), 3000); // Stops after 3 seconds

// takeWhile - emit while condition is true
of(1, 2, 3, 4, 5)
  .pipe(
    takeWhile(x => x < 4)
  )
  .subscribe(console.log); // 1, 2, 3

// skip - skip first N values
of(1, 2, 3, 4, 5)
  .pipe(
    skip(2)
  )
  .subscribe(console.log); // 3, 4, 5

// distinct - emit unique values
of(1, 2, 2, 3, 3, 4)
  .pipe(
    distinct()
  )
  .subscribe(console.log); // 1, 2, 3, 4

// distinctUntilChanged - emit when value changes from previous
of(1, 1, 2, 2, 3, 3)
  .pipe(
    distinctUntilChanged()
  )
  .subscribe(console.log); // 1, 2, 3

Combination Operators

import { merge, concat, combineLatest, forkJoin, zip } from 'rxjs';
import { delay } from 'rxjs/operators';

// merge - emit values from multiple observables as they occur
const obs1$ = of('A', 'B').pipe(delay(1000));
const obs2$ = of('1', '2').pipe(delay(500));
merge(obs1$, obs2$).subscribe(console.log); // '1', '2', 'A', 'B'

// concat - emit values sequentially (wait for previous to complete)
concat(obs1$, obs2$).subscribe(console.log); // 'A', 'B', '1', '2'

// combineLatest - emit when any observable emits (after all emit once)
const age$ = of(30, 31, 32);
const name$ = of('John', 'Jane');
combineLatest([age$, name$]).subscribe(console.log);
// [32, 'John'], [32, 'Jane']

// forkJoin - wait for all to complete, emit last values
const req1$ = of('result1').pipe(delay(1000));
const req2$ = of('result2').pipe(delay(2000));
forkJoin([req1$, req2$]).subscribe(console.log);
// ['result1', 'result2'] after 2 seconds

// zip - emit pairs of values at same index
const nums$ = of(1, 2, 3);
const chars$ = of('A', 'B', 'C');
zip(nums$, chars$).subscribe(console.log);
// [1, 'A'], [2, 'B'], [3, 'C']

Advanced Operators

switchMap, mergeMap, concatMap

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

// switchMap - cancel previous, switch to new observable
// Use for: Search, typeahead
const searchInput = document.getElementById('search');
fromEvent(searchInput, 'input')
  .pipe(
    debounceTime(300),
    distinctUntilChanged(),
    switchMap(event => {
      const query = (event.target as HTMLInputElement).value;
      return ajax.getJSON(`/api/search?q=${query}`);
    })
  )
  .subscribe(results => console.log(results));

// mergeMap - run all observables in parallel
// Use for: Independent requests
of(1, 2, 3)
  .pipe(
    mergeMap(id => ajax.getJSON(`/api/users/${id}`))
  )
  .subscribe(user => console.log(user));

// concatMap - run observables sequentially (queue)
// Use for: Ordered operations
of(1, 2, 3)
  .pipe(
    concatMap(id => ajax.post(`/api/process/${id}`, {}))
  )
  .subscribe(result => console.log(result));

// exhaustMap - ignore new observables while current is active
// Use for: Login, submit buttons
const loginButton = document.getElementById('login');
fromEvent(loginButton, 'click')
  .pipe(
    exhaustMap(() => ajax.post('/api/login', credentials))
  )
  .subscribe(response => console.log(response));

Error Handling

import { catchError, retry, retryWhen, tap } from 'rxjs/operators';
import { throwError, of, timer } from 'rxjs';

// catchError - handle errors and continue
this.http.get('/api/data')
  .pipe(
    catchError(error => {
      console.error('Error:', error);
      return of([]); // Return fallback value
    })
  )
  .subscribe(data => console.log(data));

// retry - retry on error
this.http.get('/api/data')
  .pipe(
    retry(3) // Retry up to 3 times
  )
  .subscribe();

// retryWhen - custom retry logic
this.http.get('/api/data')
  .pipe(
    retryWhen(errors =>
      errors.pipe(
        tap(error => console.log('Retrying...', error)),
        delay(1000), // Wait 1 second between retries
        take(3) // Max 3 retries
      )
    )
  )
  .subscribe();

Subjects

import { Subject, BehaviorSubject, ReplaySubject, AsyncSubject } from 'rxjs';

// Subject - no initial value, no replay
const subject = new Subject<number>();
subject.subscribe(v => console.log('Sub 1:', v));
subject.next(1); // Sub 1: 1
subject.subscribe(v => console.log('Sub 2:', v));
subject.next(2); // Sub 1: 2, Sub 2: 2

// BehaviorSubject - has initial value, replays last
const behavior = new BehaviorSubject<number>(0);
behavior.subscribe(v => console.log('Sub 1:', v)); // Sub 1: 0
behavior.next(1); // Sub 1: 1
behavior.subscribe(v => console.log('Sub 2:', v)); // Sub 2: 1

// ReplaySubject - replays last N values
const replay = new ReplaySubject<number>(2); // Buffer size 2
replay.next(1);
replay.next(2);
replay.next(3);
replay.subscribe(v => console.log('Sub:', v)); // Sub: 2, Sub: 3

// AsyncSubject - emits last value on complete
const async = new AsyncSubject<number>();
async.subscribe(v => console.log('Sub:', v));
async.next(1);
async.next(2);
async.next(3);
async.complete(); // Sub: 3