Angular — RxJS

What is RxJS?

Reactive Extensions Library for JavaScript — library for composing asynchronous code.

Creating Observables

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

// of
const numbers$ = of(1, 2, 3);

// from
const array$ = from([1, 2, 3]);

// interval
const timer$ = interval(1000);

// Subject
const subject$ = new Subject();

Transforming Operators

import { map, filter, reduce, switchMap, mergeMap, concatMap } from 'rxjs/operators';

// map
numbers$.pipe(map(x => x * 2));

// filter
numbers$.pipe(filter(x => x > 1));

// reduce
numbers$.pipe(reduce((acc, val) => acc + val, 0));

// switchMap (cancels previous)
searchControl.valueChanges.pipe(
  switchMap(term => this.http.get(`/api/search?q=${term}`))
);

// mergeMap (runs in parallel)
ids$.pipe(
  mergeMap(id => this.http.get(`/api/items/${id}`))
);

Combining Operators

import { merge, combineLatest, forkJoin } from 'rxjs';

// merge
merge(source1$, source2$);

// combineLatest
combineLatest([source1$, source2$]);

// forkJoin (waits for all to complete)
forkJoin([http1$, http2$, http3$]);

Error Handling

import { catchError, retry } from 'rxjs/operators';
import { of } from 'rxjs';

this.http.get('/api/data').pipe(
  retry(3),
  catchError(err => {
    console.error(err);
    return of([]);
  })
);

Mini Practice

1. Create observables
2. Use transformation operators
3. Combine multiple streams
4. Handle errors

Up Next

Continue with Signals — Angular signals in depth.