RxJS Operators

Creation Operators

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

// of - emit values in sequence
of(1, 2, 3).subscribe(x => console.log(x)); // 1, 2, 3

// from - convert array/promise to observable
from([1, 2, 3]).subscribe(x => console.log(x));
from(fetch('/api/data')).subscribe(response => console.log(response));

// interval - emit every N milliseconds
interval(1000).subscribe(x => console.log(x)); // 0, 1, 2, 3...

// timer - emit after delay, then interval
timer(3000, 1000).subscribe(x => console.log(x)); // Wait 3s, then emit every 1s

// fromEvent - convert event to observable
const clicks$ = fromEvent(document, 'click');
clicks$.subscribe(event => console.log(event));

Transformation Operators

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

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

// mapTo - map to constant value
clicks$.pipe(
  mapTo(1)
).subscribe(x => console.log(x)); // 1, 1, 1...

// scan - accumulate values (like reduce)
of(1, 2, 3).pipe(
  scan((acc, value) => acc + value, 0)
).subscribe(x => console.log(x)); // 1, 3, 6

Filtering Operators

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

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

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

// takeUntil - take until another observable emits
const stop$ = new Subject();
interval(1000).pipe(
  takeUntil(stop$)
).subscribe(x => console.log(x));

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

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

// distinct - emit only unique values
of(1, 1, 2, 2, 3).pipe(
  distinct()
).subscribe(x => console.log(x)); // 1, 2, 3

// debounceTime - emit after silence period
searchInput$.pipe(
  debounceTime(300)
).subscribe(value => this.search(value));

Combination Operators

import { merge, concat, combineLatest, forkJoin, zip } from 'rxjs';
import { mergeMap, concatMap, switchMap, exhaustMap } from 'rxjs/operators';

// merge - emit values from multiple observables
merge(
  of(1, 2, 3),
  of(4, 5, 6)
).subscribe(x => console.log(x)); // 1, 2, 3, 4, 5, 6

// concat - emit one observable after another
concat(
  of(1, 2, 3),
  of(4, 5, 6)
).subscribe(x => console.log(x)); // 1, 2, 3, 4, 5, 6

// combineLatest - emit when any observable emits
combineLatest([
  of('A', 'B'),
  of(1, 2)
]).subscribe(([letter, number]) => {
  console.log(letter, number); // B 1, B 2
});

// forkJoin - wait for all to complete
forkJoin({
  users: this.http.get('/users'),
  posts: this.http.get('/posts')
}).subscribe(({ users, posts }) => {
  console.log(users, posts);
});

// zip - combine values by index
zip(
  of('A', 'B', 'C'),
  of(1, 2, 3)
).subscribe(([letter, number]) => {
  console.log(letter, number); // A 1, B 2, C 3
});

Flattening Operators

// switchMap - cancel previous, switch to new observable
searchTerm$.pipe(
  debounceTime(300),
  switchMap(term => this.http.get(`/search?q=${term}`))
).subscribe(results => console.log(results));

// mergeMap - run all concurrently
users$.pipe(
  mergeMap(user => this.http.get(`/posts/${user.id}`))
).subscribe(posts => console.log(posts));

// concatMap - run sequentially
requests$.pipe(
  concatMap(req => this.http.post('/api', req))
).subscribe(response => console.log(response));

// exhaustMap - ignore new values while current is active
saveButton$.pipe(
  exhaustMap(() => this.http.post('/save', data))
).subscribe(response => console.log(response));

Error Handling

import { catchError, retry, retryWhen, throwError } from 'rxjs';

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

// retry - retry N times on error
this.http.get('/api/data').pipe(
  retry(3),
  catchError(error => throwError(() => error))
).subscribe();

// retryWhen - custom retry logic
this.http.get('/api/data').pipe(
  retryWhen(errors =>
    errors.pipe(
      delay(1000),
      take(3)
    )
  )
).subscribe();

Utility Operators

import { tap, delay, timeout, finalize } from 'rxjs/operators';

// tap - side effects (debugging, logging)
this.http.get('/api/data').pipe(
  tap(data => console.log('Data:', data)),
  map(data => data.items)
).subscribe();

// delay - delay emissions
of(1, 2, 3).pipe(
  delay(1000)
).subscribe(x => console.log(x)); // After 1 second

// timeout - error if no emission within time
this.http.get('/api/data').pipe(
  timeout(5000), // 5 seconds
  catchError(error => of([]))
).subscribe();

// finalize - run code when observable completes/errors
this.http.get('/api/data').pipe(
  finalize(() => this.loading = false)
).subscribe();

Common Patterns

Search with Debounce

searchTerm$ = new Subject<string>();

ngOnInit() {
  this.searchTerm$.pipe(
    debounceTime(300),
    distinctUntilChanged(),
    switchMap(term => this.searchService.search(term))
  ).subscribe(results => {
    this.results = results;
  });
}

onSearch(term: string) {
  this.searchTerm$.next(term);
}

Unsubscribe Pattern

private destroy$ = new Subject<void>();

ngOnInit() {
  this.dataService.getData().pipe(
    takeUntil(this.destroy$)
  ).subscribe(data => this.data = data);
}

ngOnDestroy() {
  this.destroy$.next();
  this.destroy$.complete();
}

Loading State

loadData() {
  this.loading = true;

  this.http.get('/api/data').pipe(
    finalize(() => this.loading = false)
  ).subscribe(
    data => this.data = data,
    error => this.error = error
  );
}

Cache

private cache$ = new Map<string, Observable<any>>();

getData(id: string): Observable<any> {
  if (!this.cache$.has(id)) {
    this.cache$.set(
      id,
      this.http.get(`/api/data/${id}`).pipe(shareReplay(1))
    );
  }
  return this.cache$.get(id)!;
}

Best Practices

• Use switchMap for searches (cancel previous)
• Use mergeMap for independent requests
• Use concatMap for sequential operations
• Always handle errors with catchError
• Use takeUntil to unsubscribe
• Use shareReplay for caching
• Use debounceTime for user input
• Use distinctUntilChanged to avoid duplicates
• Avoid nested subscriptions (use flattening operators)