RxJS Cheat Sheet

RxJS Cheat Sheet

Complete reference for RxJS operators, observables, and reactive patterns — from creation to error handling, with practical examples.

11
Sections
60+
Examples
RxJS 7+
Version
Patterns
Real-World

Observable Creation

Create observables from various data sources.

Beginner
7 examples

of RxJS

Emit a sequence of values synchronously.

import { of } from 'rxjs';

// Emit values 1, 2, 3 then complete
const source$ = of(1, 2, 3);

source$.subscribe({
  next: val => console.log(val),      // 1, 2, 3
  complete: () => console.log('Done')
});

from RxJS

Convert arrays, promises, or iterables to observables.

import { from } from 'rxjs';

// From array
const arr$ = from([1, 2, 3]);
arr$.subscribe(console.log); // 1, 2, 3

// From Promise
const promise$ = from(fetch('/api/data').then(r => r.json()));
promise$.subscribe(data => console.log(data));

// From iterable
const set$ = from(new Set([1, 2, 3]));
set$.subscribe(console.log);

interval & timer RxJS

Emit sequential numbers at specified intervals.

import { interval, timer } from 'rxjs';
import { take } from 'rxjs/operators';

// Emit every second: 0, 1, 2, 3...
const interval$ = interval(1000);
interval$.pipe(take(5)).subscribe(console.log);

// Wait 2s, then emit every 1s
const timer$ = timer(2000, 1000);
timer$.pipe(take(3)).subscribe(console.log);

// Single emission after delay
const delayed$ = timer(3000);
delayed$.subscribe(() => console.log('3 seconds passed'));

fromEvent RxJS

Create observable from DOM events.

import { fromEvent } from 'rxjs';
import { map, throttleTime } from 'rxjs/operators';

// Click events
const clicks$ = fromEvent(document, 'click');
clicks$.subscribe(event => console.log('Clicked:', event.clientX, event.clientY));

// Keyboard events with mapping
const keypress$ = fromEvent<KeyboardEvent>(document, 'keydown');
keypress$
  .pipe(
    map(e => e.key),
    throttleTime(300)
  )
  .subscribe(key => console.log('Key:', key));

Observable Constructor RxJS

Create custom observables with full control.

import { Observable } from 'rxjs';

const custom$ = new Observable(subscriber => {
  subscriber.next(1);
  subscriber.next(2);
  
  // Async emission
  const timeoutId = setTimeout(() => {
    subscriber.next(3);
    subscriber.complete();
  }, 1000);
  
  // Cleanup function (called on unsubscribe)
  return () => {
    console.log('Cleanup');
    clearTimeout(timeoutId);
  };
});

const subscription = custom$.subscribe({
  next: val => console.log(val),
  complete: () => console.log('Complete')
});

// Unsubscribe after 500ms (triggers cleanup)
setTimeout(() => subscription.unsubscribe(), 500);

defer RxJS

Create observable lazily on subscription.

import { defer, of } from 'rxjs';

// Factory function called on each subscription
const deferred$ = defer(() => {
  const random = Math.random();
  console.log('Creating observable with:', random);
  return of(random);
});

deferred$.subscribe(console.log); // Different value
deferred$.subscribe(console.log); // Different value each time

EMPTY, NEVER, throwError RxJS

Special utility observables.

import { EMPTY, NEVER, throwError } from 'rxjs';

// Completes immediately, no emissions
EMPTY.subscribe({
  complete: () => console.log('Completed immediately')
});

// Never emits, never completes
NEVER.subscribe({
  next: () => console.log('Never called'),
  complete: () => console.log('Never called')
});

// Immediately errors
throwError(() => new Error('Something went wrong'))
  .subscribe({
    error: err => console.error('Error:', err.message)
  });

Transformation Operators

Transform emitted values into new values or observables.

Beginner
8 examples

map RxJS

Transform each emitted value.

import { of } from 'rxjs';
import { map } from 'rxjs/operators';

const source$ = of(1, 2, 3, 4, 5);

source$.pipe(
  map(x => x * 10)
).subscribe(console.log); // 10, 20, 30, 40, 50

// With index
source$.pipe(
  map((value, index) => ({ value, index }))
).subscribe(console.log); // { value: 1, index: 0 }, ...

switchMap RxJS

Map to observable, cancel previous inner observable on new emission.

import { fromEvent, interval } from 'rxjs';
import { switchMap, take } from 'rxjs/operators';

// Typeahead search - cancels previous request
const search$ = fromEvent(input, 'input').pipe(
  switchMap(e => fetch(`/api/search?q=${e.target.value}\`).then(r => r.json()))
);

// Each click restarts the interval
fromEvent(document, 'click').pipe(
  switchMap(() => interval(1000).pipe(take(5)))
).subscribe(console.log); // Resets to 0 on each click

mergeMap (flatMap) RxJS

Map to observable, merge all inner observables concurrently.

import { of, interval } from 'rxjs';
import { mergeMap, take, delay } from 'rxjs/operators';

// Process all requests concurrently
const ids$ = of(1, 2, 3);

ids$.pipe(
  mergeMap(id => 
    of(`Result for ${id}\`).pipe(delay(Math.random() * 1000))
  )
).subscribe(console.log); // Results in random order

// With concurrency limit
ids$.pipe(
  mergeMap(id => fetchUser(id), 2) // Max 2 concurrent
).subscribe(console.log);

concatMap RxJS

Map to observable, process inner observables sequentially.

import { of } from 'rxjs';
import { concatMap, delay } from 'rxjs/operators';

const ids$ = of(1, 2, 3);

// Process one at a time, maintain order
ids$.pipe(
  concatMap(id => 
    of(`Result for `).pipe(delay(1000))
  )
).subscribe(console.log);
// After 1s: Result for 1
// After 2s: Result for 2
// After 3s: Result for 3

exhaustMap RxJS

Ignore new values while inner observable is active.

import { fromEvent, interval } from 'rxjs';
import { exhaustMap, take } from 'rxjs/operators';

// Prevent double-click submissions
const submit$ = fromEvent(submitBtn, 'click').pipe(
  exhaustMap(() => 
    saveData().pipe(take(1)) // Ignore clicks until save completes
  )
);

// Button clicks during 3s interval are ignored
fromEvent(document, 'click').pipe(
  exhaustMap(() => interval(1000).pipe(take(3)))
).subscribe(console.log);

scan RxJS

Accumulate values over time (like reduce but emits each step).

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

// Running total
of(1, 2, 3, 4, 5).pipe(
  scan((acc, val) => acc + val, 0)
).subscribe(console.log); // 1, 3, 6, 10, 15

// Click counter
fromEvent(document, 'click').pipe(
  scan(count => count + 1, 0)
).subscribe(count => console.log('Clicks:', count));

// Build array over time
of('a', 'b', 'c').pipe(
  scan((arr, val) => [...arr, val], [])
).subscribe(console.log); // ['a'], ['a','b'], ['a','b','c']

pluck & mapTo RxJS

Extract properties or map to constant values.

import { fromEvent, interval } from 'rxjs';
import { pluck, map } from 'rxjs/operators';

// Extract nested property (deprecated, use map instead)
fromEvent(document, 'click').pipe(
  map(event => event.target.tagName) // Modern approach
).subscribe(console.log);

// Map to constant (deprecated, use map instead)
interval(1000).pipe(
  map(() => 'tick') // Modern approach
).subscribe(console.log); // 'tick', 'tick', ...

buffer & bufferTime RxJS

Collect emitted values into arrays.

import { interval, fromEvent } from 'rxjs';
import { buffer, bufferTime, bufferCount } from 'rxjs/operators';

// Buffer until click
interval(500).pipe(
  buffer(fromEvent(document, 'click'))
).subscribe(arr => console.log('Buffered:', arr));

// Buffer every 2 seconds
interval(300).pipe(
  bufferTime(2000)
).subscribe(arr => console.log('2s buffer:', arr));

// Buffer every 3 emissions
interval(500).pipe(
  bufferCount(3)
).subscribe(arr => console.log('Count buffer:', arr));

Filtering Operators

Select or limit emitted values based on conditions.

Beginner
8 examples

filter RxJS

Emit only values that pass a predicate.

import { of, from } from 'rxjs';
import { filter } from 'rxjs/operators';

of(1, 2, 3, 4, 5, 6).pipe(
  filter(x => x % 2 === 0)
).subscribe(console.log); // 2, 4, 6

// With type guard
interface User { name: string; active: boolean }
const users$: Observable<User | null> = from([...]);

users$.pipe(
  filter((user): user is User => user !== null && user.active)
).subscribe(user => console.log(user.name)); // TypeScript knows user is User

take & takeLast RxJS

Limit emissions to first/last N values.

import { of, interval } from 'rxjs';
import { take, takeLast } from 'rxjs/operators';

// Take first 3
interval(1000).pipe(
  take(3)
).subscribe(console.log); // 0, 1, 2 then complete

// Take last 2 (waits for complete)
of(1, 2, 3, 4, 5).pipe(
  takeLast(2)
).subscribe(console.log); // 4, 5

first & last RxJS

Emit only the first/last value matching a condition.

import { of, EMPTY } from 'rxjs';
import { first, last } from 'rxjs/operators';

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

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

of(1, 2, 3, 4, 5).pipe(
  last()
).subscribe(console.log); // 5

// With default value (no error if empty)
EMPTY.pipe(
  first(null, 'default')
).subscribe(console.log); // 'default'

takeUntil & takeWhile RxJS

Take values until a condition or notifier.

import { interval, fromEvent, of } from 'rxjs';
import { takeUntil, takeWhile } from 'rxjs/operators';

// Take until another observable emits
const stop$ = fromEvent(document, 'click');

interval(500).pipe(
  takeUntil(stop$)
).subscribe(console.log); // Stops on first click

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

// Include the failing value
of(1, 2, 3, 4, 5).pipe(
  takeWhile(x => x < 4, true) // inclusive
).subscribe(console.log); // 1, 2, 3, 4

skip & skipUntil RxJS

Skip initial emissions.

import { of, interval, timer } from 'rxjs';
import { skip, skipUntil, skipWhile } from 'rxjs/operators';

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

// Skip until timer
interval(500).pipe(
  skipUntil(timer(2000)),
  take(3)
).subscribe(console.log); // Starts after 2s

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

debounceTime RxJS

Emit only after a pause in emissions.

import { fromEvent } from 'rxjs';
import { debounceTime, map } from 'rxjs/operators';

// Search input - wait 300ms after user stops typing
const search$ = fromEvent(searchInput, 'input').pipe(
  debounceTime(300),
  map(e => e.target.value)
);

search$.subscribe(query => {
  console.log('Searching for:', query);
  // Make API call here
});

throttleTime RxJS

Emit first value, then ignore for duration.

import { fromEvent } from 'rxjs';
import { throttleTime } from 'rxjs/operators';

// Rate limit scroll events
fromEvent(window, 'scroll').pipe(
  throttleTime(200)
).subscribe(() => {
  console.log('Scroll position:', window.scrollY);
});

// Prevent rapid button clicks
fromEvent(button, 'click').pipe(
  throttleTime(1000)
).subscribe(() => {
  console.log('Button clicked (throttled)');
});

distinctUntilChanged RxJS

Only emit when value changes from previous.

import { of } from 'rxjs';
import { distinctUntilChanged } from 'rxjs/operators';

of(1, 1, 2, 2, 2, 3, 1, 1).pipe(
  distinctUntilChanged()
).subscribe(console.log); // 1, 2, 3, 1

// With custom comparison
const users$ = of(
  { id: 1, name: 'Alice' },
  { id: 1, name: 'Alice Updated' },
  { id: 2, name: 'Bob' }
);

users$.pipe(
  distinctUntilChanged((prev, curr) => prev.id === curr.id)
).subscribe(console.log); // First Alice, Bob (skips updated Alice)

Combination Operators

Combine multiple observables into one.

Intermediate
7 examples

combineLatest RxJS

Emit array of latest values when any source emits.

import { combineLatest, interval, of } from 'rxjs';
import { map, take } from 'rxjs/operators';

const timer1$ = interval(1000).pipe(take(3));
const timer2$ = interval(1500).pipe(take(3));

combineLatest([timer1$, timer2$]).pipe(
  map(([t1, t2]) => `Timer1: , Timer2: `)
).subscribe(console.log);
// Emits when either updates (after both have emitted once)

// Form validation example
const username$ = usernameInput.valueChanges;
const password$ = passwordInput.valueChanges;

combineLatest([username$, password$]).pipe(
  map(([user, pass]) => user.length > 0 && pass.length >= 8)
).subscribe(isValid => submitBtn.disabled = !isValid);

merge RxJS

Merge multiple observables into one, emitting all values.

import { merge, interval, fromEvent } from 'rxjs';
import { map, take } from 'rxjs/operators';

const clicks$ = fromEvent(document, 'click').pipe(map(() => 'click'));
const keys$ = fromEvent(document, 'keydown').pipe(map(() => 'keydown'));

merge(clicks$, keys$).subscribe(console.log);
// Emits 'click' or 'keydown' as events occur

// Merge intervals
const fast$ = interval(500).pipe(map(n => `Fast: `), take(5));
const slow$ = interval(1000).pipe(map(n => `Slow: `), take(3));

merge(fast$, slow$).subscribe(console.log);

concat RxJS

Subscribe to observables sequentially, one after another.

import { concat, of, interval } from 'rxjs';
import { take, delay } from 'rxjs/operators';

const first$ = of(1, 2, 3);
const second$ = of(4, 5, 6);
const third$ = interval(500).pipe(take(3));

concat(first$, second$, third$).subscribe(console.log);
// 1, 2, 3, 4, 5, 6, 0, 1, 2 (in sequence)

// Retry with delay
const fetchData$ = httpRequest$.pipe(
  catchError(() => concat(
    of(null).pipe(delay(1000)), // Wait 1 second
    fetchData$ // Retry
  ))
);

forkJoin RxJS

Wait for all observables to complete, emit final values.

import { forkJoin, of, timer } from 'rxjs';
import { map } from 'rxjs/operators';

// Parallel API requests
const user$ = fetch('/api/user').then(r => r.json());
const posts$ = fetch('/api/posts').then(r => r.json());
const comments$ = fetch('/api/comments').then(r => r.json());

forkJoin({
  user: user$,
  posts: posts$,
  comments: comments$
}).subscribe(result => {
  console.log(result.user);
  console.log(result.posts);
  console.log(result.comments);
});

// ⚠️ If any observable errors, forkJoin errors
// ⚠️ If any observable never completes, forkJoin never emits

zip RxJS

Combine corresponding values from multiple observables.

import { zip, of, interval } from 'rxjs';
import { take, map } from 'rxjs/operators';

const a$ = of(1, 2, 3);
const b$ = of('a', 'b', 'c');
const c$ = of(true, false, true);

zip(a$, b$, c$).subscribe(console.log);
// [1, 'a', true]
// [2, 'b', false]
// [3, 'c', true]

// Zip with interval creates delay
const numbers$ = of(1, 2, 3, 4);
const delay$ = interval(1000);

zip(numbers$, delay$).pipe(
  map(([num]) => num)
).subscribe(console.log); // 1, 2, 3, 4 (one per second)

withLatestFrom RxJS

Combine source with latest from other observables.

import { fromEvent, interval } from 'rxjs';
import { withLatestFrom, map } from 'rxjs/operators';

const click$ = fromEvent(document, 'click');
const timer$ = interval(1000);

click$.pipe(
  withLatestFrom(timer$),
  map(([event, timerVal]) => `Clicked at timer: `)
).subscribe(console.log);
// Only emits on click, with latest timer value

// Form submission with current form state
submitBtn$.pipe(
  withLatestFrom(formState$),
  map(([_, formData]) => formData)
).subscribe(data => saveData(data));

race RxJS

Use the first observable to emit.

import { race, timer, of } from 'rxjs';
import { map, delay } from 'rxjs/operators';

const fast$ = timer(100).pipe(map(() => 'Fast wins!'));
const slow$ = timer(500).pipe(map(() => 'Slow loses'));

race(fast$, slow$).subscribe(console.log); // 'Fast wins!'

// Timeout pattern
const request$ = fetch('/api/data').then(r => r.json());
const timeout$ = timer(5000).pipe(map(() => { throw new Error('Timeout'); }));

race(request$, timeout$).subscribe({
  next: data => console.log(data),
  error: err => console.error(err.message)
});

Error Handling

Handle errors gracefully in observable streams.

Intermediate
4 examples

catchError RxJS

Handle errors and return a fallback observable.

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

const source$ = throwError(() => new Error('Oops!'));

source$.pipe(
  catchError(err => {
    console.error('Caught:', err.message);
    return of('Fallback value'); // Return fallback
  })
).subscribe(console.log); // 'Fallback value'

// Rethrow with additional context
fetchData$.pipe(
  catchError(err => {
    return throwError(() => new Error(`Fetch failed: ${err.message}\`));
  })
);

// Return empty to swallow error
source$.pipe(
  catchError(() => EMPTY)
).subscribe(); // No error, no emissions

retry & retryWhen RxJS

Automatically retry failed observables.

import { of, throwError, timer } from 'rxjs';
import { retry, retryWhen, delay, take, concatMap } from 'rxjs/operators';

let attempt = 0;
const flaky$ = of('try').pipe(
  concatMap(() => {
    attempt++;
    if (attempt < 3) throw new Error('Fail');
    return of('Success!');
  })
);

// Retry up to 2 times
flaky$.pipe(
  retry(2)
).subscribe(console.log);

// Retry with delay (using retry config in RxJS 7+)
failingRequest$.pipe(
  retry({
    count: 3,
    delay: (error, retryCount) => {
      console.log(`Retry ${retryCount}\ after error: ${error.message}\`);
      return timer(retryCount * 1000); // Exponential backoff
    }
  })
).subscribe();

finalize RxJS

Execute cleanup logic on complete or error.

import { of, throwError } from 'rxjs';
import { finalize, delay } from 'rxjs/operators';

// Always runs, like try/finally
of('data').pipe(
  delay(1000),
  finalize(() => {
    console.log('Cleanup: hide loading spinner');
  })
).subscribe(console.log);

// Runs on error too
throwError(() => new Error('Fail')).pipe(
  finalize(() => console.log('Cleanup on error'))
).subscribe({
  error: () => console.log('Error handler')
});
// Output: 'Cleanup on error', 'Error handler'

timeout & timeoutWith RxJS

Error or switch if observable takes too long.

import { of, timer } from 'rxjs';
import { timeout, delay, catchError } from 'rxjs/operators';

// Error after 2 seconds
const slow$ = of('result').pipe(delay(5000));

slow$.pipe(
  timeout(2000),
  catchError(err => {
    console.log('Timed out!');
    return of('Default value');
  })
).subscribe(console.log);

// Timeout with config (RxJS 7+)
slow$.pipe(
  timeout({
    each: 2000, // Timeout between emissions
    with: () => of('Fallback') // Switch to this on timeout
  })
).subscribe(console.log);

Subjects

Special observables that are also observers — multicast values.

Intermediate
5 examples

Subject RxJS

Basic subject — no initial value, no replay.

import { Subject } from 'rxjs';

const subject$ = new Subject<number>();

// Subscriber A
subject$.subscribe(val => console.log('A:', val));

subject$.next(1); // A: 1

// Subscriber B (late)
subject$.subscribe(val => console.log('B:', val));

subject$.next(2); // A: 2, B: 2

subject$.complete();

BehaviorSubject RxJS

Holds current value, new subscribers get latest immediately.

import { BehaviorSubject } from 'rxjs';

const user$ = new BehaviorSubject<string>('Guest');

// Get current value synchronously
console.log('Current:', user$.getValue()); // 'Guest'

// New subscriber gets current value immediately
user$.subscribe(name => console.log('User:', name)); // 'User: Guest'

user$.next('Alice'); // 'User: Alice'

// Late subscriber still gets latest
user$.subscribe(name => console.log('Late:', name)); // 'Late: Alice'

// Common pattern: state management
const state$ = new BehaviorSubject({ count: 0, loading: false });
state$.next({ ...state$.getValue(), loading: true });

ReplaySubject RxJS

Replay N previous values to new subscribers.

import { ReplaySubject } from 'rxjs';

// Replay last 2 values
const replay$ = new ReplaySubject<number>(2);

replay$.next(1);
replay$.next(2);
replay$.next(3);

// New subscriber gets 2, 3 (last 2 values)
replay$.subscribe(val => console.log('Replayed:', val));
// Output: 2, 3

// With time window: replay values from last 500ms
const timedReplay$ = new ReplaySubject<number>(100, 500);

// Replay all values (use Infinity)
const replayAll$ = new ReplaySubject<number>(Infinity);

AsyncSubject RxJS

Only emits the last value, and only when completed.

import { AsyncSubject } from 'rxjs';

const async$ = new AsyncSubject<number>();

async$.subscribe(val => console.log('Received:', val));

async$.next(1);
async$.next(2);
async$.next(3);
// Nothing logged yet

async$.complete();
// Now logs: 'Received: 3' (only the last value)

// Useful for caching single results
const cachedRequest$ = new AsyncSubject();
fetch('/api/data')
  .then(r => r.json())
  .then(data => {
    cachedRequest$.next(data);
    cachedRequest$.complete();
  });

Subject as Observer RxJS

Use subject to multicast a source observable.

import { Subject, interval } from 'rxjs';
import { take } from 'rxjs/operators';

const source$ = interval(1000).pipe(take(3));
const subject$ = new Subject<number>();

// Multiple subscribers share the same execution
subject$.subscribe(val => console.log('A:', val));
subject$.subscribe(val => console.log('B:', val));

// Pipe source into subject
source$.subscribe(subject$);
// Both A and B receive: 0, 1, 2

// This is essentially what share() does automatically

Multicasting & Sharing

Share observable executions among multiple subscribers.

Advanced
4 examples

share RxJS

Share source among subscribers, resubscribe when all unsubscribe.

import { interval } from 'rxjs';
import { share, take, tap } from 'rxjs/operators';

const source$ = interval(1000).pipe(
  tap(val => console.log('Source emits:', val)),
  take(5),
  share() // Share execution
);

// Both subscribers receive same values
source$.subscribe(val => console.log('A:', val));
setTimeout(() => {
  source$.subscribe(val => console.log('B:', val));
}, 2500);
// B joins late, gets values from that point on

shareReplay RxJS

Share and replay N values to late subscribers.

import { of, defer } from 'rxjs';
import { shareReplay, delay, tap } from 'rxjs/operators';

// Cache API response
const cachedData$ = defer(() => {
  console.log('Fetching data...');
  return fetch('/api/data').then(r => r.json());
}).pipe(
  shareReplay(1) // Cache the result
);

// First subscription triggers fetch
cachedData$.subscribe(data => console.log('First:', data));

// Second subscription uses cache
setTimeout(() => {
  cachedData$.subscribe(data => console.log('Second:', data));
}, 1000);
// 'Fetching data...' only logged once

publish & refCount RxJS

Manual control over multicasting.

import { interval } from 'rxjs';
import { publish, refCount, take } from 'rxjs/operators';

// publish() creates a ConnectableObservable
const source$ = interval(500).pipe(
  take(5),
  publish(),
  refCount() // Auto-connect when first subscriber, disconnect when last leaves
);

const sub1 = source$.subscribe(val => console.log('A:', val));
const sub2 = source$.subscribe(val => console.log('B:', val));

// Both receive same values

setTimeout(() => {
  sub1.unsubscribe();
  sub2.unsubscribe();
  // Source stops because refCount drops to 0
}, 2000);

connect (RxJS 7+) RxJS

Modern approach to multicasting with connect.

import { interval, connectable, Subject } from 'rxjs';
import { take } from 'rxjs/operators';

const source$ = interval(500).pipe(take(5));

// Create connectable observable
const multicasted$ = connectable(source$, {
  connector: () => new Subject(),
  resetOnDisconnect: true
});

multicasted$.subscribe(val => console.log('A:', val));
multicasted$.subscribe(val => console.log('B:', val));

// Manually connect
const connection = multicasted$.connect();

// Disconnect after 2 seconds
setTimeout(() => connection.unsubscribe(), 2000);

Utility Operators

Helpful operators for debugging, timing, and more.

Beginner
6 examples

tap RxJS

Perform side effects without modifying the stream.

import { of } from 'rxjs';
import { tap, map, filter } from 'rxjs/operators';

of(1, 2, 3, 4, 5).pipe(
  tap(val => console.log('Before filter:', val)),
  filter(x => x % 2 === 0),
  tap(val => console.log('After filter:', val)),
  map(x => x * 10),
  tap({
    next: val => console.log('Final:', val),
    complete: () => console.log('Done!')
  })
).subscribe();

// Great for debugging pipelines

delay & delayWhen RxJS

Delay emissions by time or observable.

import { of, timer } from 'rxjs';
import { delay, delayWhen } from 'rxjs/operators';

// Fixed delay
of('Hello').pipe(
  delay(2000)
).subscribe(console.log); // After 2 seconds

// Dynamic delay per value
of(1, 2, 3).pipe(
  delayWhen(val => timer(val * 1000))
).subscribe(console.log);
// 1 after 1s, 2 after 2s, 3 after 3s

toArray RxJS

Collect all emissions into an array on complete.

import { of, interval } from 'rxjs';
import { toArray, take, filter } from 'rxjs/operators';

of(1, 2, 3, 4, 5).pipe(
  filter(x => x % 2 === 0),
  toArray()
).subscribe(console.log); // [2, 4]

interval(100).pipe(
  take(5),
  toArray()
).subscribe(console.log); // [0, 1, 2, 3, 4]

startWith & endWith RxJS

Prepend or append values to stream.

import { of } from 'rxjs';
import { startWith, endWith } from 'rxjs/operators';

of(2, 3).pipe(
  startWith(1),
  endWith(4)
).subscribe(console.log); // 1, 2, 3, 4

// Common: loading state
dataRequest$.pipe(
  map(data => ({ loading: false, data })),
  startWith({ loading: true, data: null })
).subscribe(state => console.log(state));

defaultIfEmpty RxJS

Emit default value if source completes empty.

import { EMPTY, of } from 'rxjs';
import { defaultIfEmpty, filter } from 'rxjs/operators';

EMPTY.pipe(
  defaultIfEmpty('Nothing here')
).subscribe(console.log); // 'Nothing here'

of(1, 2, 3).pipe(
  filter(x => x > 10),
  defaultIfEmpty(-1)
).subscribe(console.log); // -1 (no values passed filter)

observeOn & subscribeOn RxJS

Control scheduler for emissions/subscription.

import { of, asyncScheduler, asapScheduler } from 'rxjs';
import { observeOn, subscribeOn, tap } from 'rxjs/operators';

console.log('Start');

of(1, 2, 3).pipe(
  tap(val => console.log('Emitting:', val)),
  observeOn(asyncScheduler) // Emissions on async scheduler
).subscribe(val => console.log('Received:', val));

console.log('End');

// Output order:
// Start
// End
// Emitting: 1
// Received: 1
// Emitting: 2
// Received: 2
// ...

Higher-Order Patterns

Common patterns for managing complex streams.

Advanced
5 examples

Flattening Strategy Comparison RxJS

Choose the right flattening operator for your use case.

// mergeMap: All inner observables run concurrently
// Use for: Independent parallel requests
clicks$.pipe(mergeMap(e => saveClick(e)));

// switchMap: Cancel previous, switch to new
// Use for: Typeahead, route changes
input$.pipe(switchMap(query => search(query)));

// concatMap: Queue and process one at a time  
// Use for: Sequential operations, ordered writes
ids$.pipe(concatMap(id => processInOrder(id)));

// exhaustMap: Ignore new while busy
// Use for: Prevent double submits, ignore during animation
submit$.pipe(exhaustMap(() => saveData()));

Nested Observables RxJS

Handle observables that emit observables.

import { of, interval } from 'rxjs';
import { map, mergeAll, switchAll, concatAll } from 'rxjs/operators';

// Higher-order observable (emits observables)
const higher$ = of(1, 2, 3).pipe(
  map(n => interval(1000).pipe(take(3), map(i => `-`)))
);

// Flatten with mergeAll (concurrent)
higher$.pipe(mergeAll()).subscribe(console.log);

// Flatten with switchAll (only latest)
higher$.pipe(switchAll()).subscribe(console.log);

// Flatten with concatAll (sequential)
higher$.pipe(concatAll()).subscribe(console.log);

Pagination Pattern RxJS

Handle paginated API requests.

import { BehaviorSubject, of, EMPTY } from 'rxjs';
import { expand, takeWhile, reduce, switchMap } from 'rxjs/operators';

interface Page<T> { data: T[]; nextPage: number | null; }

function fetchPage(page: number): Observable<Page<User>> {
  return fetch(`/api/users?page=`).then(r => r.json());
}

// Fetch all pages
function fetchAllPages() {
  return fetchPage(1).pipe(
    expand(response => 
      response.nextPage ? fetchPage(response.nextPage) : EMPTY
    ),
    reduce((all, page) => [...all, ...page.data], [] as User[])
  );
}

// Load more pattern
const page$ = new BehaviorSubject(1);
const users$ = page$.pipe(
  switchMap(page => fetchPage(page))
);

function loadMore() {
  page$.next(page$.getValue() + 1);
}

Polling Pattern RxJS

Repeatedly fetch data at intervals.

import { timer, of, EMPTY } from 'rxjs';
import { switchMap, takeUntil, retry, catchError, tap } from 'rxjs/operators';

const stopPolling$ = new Subject<void>();

// Basic polling every 5 seconds
timer(0, 5000).pipe(
  switchMap(() => fetchData()),
  takeUntil(stopPolling$)
).subscribe(data => updateUI(data));

// Smart polling with error handling and backoff
function pollWithRetry(intervalMs: number) {
  return timer(0, intervalMs).pipe(
    switchMap(() => fetchData().pipe(
      retry({ count: 3, delay: 1000 }),
      catchError(err => {
        console.error('Poll failed:', err);
        return EMPTY; // Continue polling
      })
    )),
    takeUntil(stopPolling$)
  );
}

// Stop polling
stopPolling$.next();

Caching Pattern RxJS

Cache responses with optional expiration.

import { of, timer, Observable } from 'rxjs';
import { shareReplay, switchMap, startWith, map } from 'rxjs/operators';

// Simple cache (never expires)
const cachedUsers$ = fetchUsers().pipe(
  shareReplay(1)
);

// Cache with expiration
function cachedWithExpiry<T>(
  source$: Observable<T>, 
  expiryMs: number
): Observable<T> {
  let cache$: Observable<T> | null = null;
  let lastFetch = 0;
  
  return new Observable(subscriber => {
    const now = Date.now();
    if (!cache$ || now - lastFetch > expiryMs) {
      lastFetch = now;
      cache$ = source$.pipe(shareReplay(1));
    }
    return cache$.subscribe(subscriber);
  });
}

// Usage
const users$ = cachedWithExpiry(fetchUsers(), 60000); // 1 min cache

Testing RxJS

Test observables with marble testing and helpers.

Advanced
4 examples

Marble Testing Basics RxJS

Use marble diagrams to test observable timing.

import { TestScheduler } from 'rxjs/testing';
import { map, delay } from 'rxjs/operators';

describe('Marble Tests', () => {
  let scheduler: TestScheduler;

  beforeEach(() => {
    scheduler = new TestScheduler((actual, expected) => {
      expect(actual).toEqual(expected);
    });
  });

  it('should double values', () => {
    scheduler.run(({ cold, expectObservable }) => {
      const source$ = cold('  -a-b-c|', { a: 1, b: 2, c: 3 });
      const expected =      '-a-b-c|';
      const result$ = source$.pipe(map(x => x * 2));
      
      expectObservable(result$).toBe(expected, { a: 2, b: 4, c: 6 });
    });
  });
});

Testing Async Operations RxJS

Handle time-based operators in tests.

import { TestScheduler } from 'rxjs/testing';
import { debounceTime, switchMap, delay } from 'rxjs/operators';

it('should debounce input', () => {
  scheduler.run(({ cold, expectObservable }) => {
    // 'a' at 0ms, 'b' at 10ms, 'c' at 50ms
    const source$ = cold('ab 40ms c|');
    // Only 'c' passes (20ms debounce)
    const expected =     '-- 40ms 20ms c|';
    
    const result$ = source$.pipe(debounceTime(20));
    expectObservable(result$).toBe(expected);
  });
});

it('should handle switchMap', () => {
  scheduler.run(({ cold, hot, expectObservable }) => {
    const trigger$ = hot('--a------b------|');
    const inner$ =       cold('  ---x|');
    
    // 'b' cancels the first inner, starts new
    const expected =     '-----x------x---|';
    
    const result$ = trigger$.pipe(
      switchMap(() => inner$)
    );
    expectObservable(result$).toBe(expected);
  });
});

Integration Testing RxJS

Test observables without marble diagrams.

import { of, firstValueFrom, lastValueFrom } from 'rxjs';
import { delay, toArray, take } from 'rxjs/operators';

describe('Integration Tests', () => {
  it('should emit correct values', async () => {
    const source$ = of(1, 2, 3).pipe(
      map(x => x * 2)
    );
    
    const result = await lastValueFrom(source$.pipe(toArray()));
    expect(result).toEqual([2, 4, 6]);
  });

  it('should handle async operations', async () => {
    const source$ = of('data').pipe(delay(100));
    
    const result = await firstValueFrom(source$);
    expect(result).toBe('data');
  });

  it('should work with done callback', (done) => {
    const values: number[] = [];
    
    of(1, 2, 3).subscribe({
      next: val => values.push(val),
      complete: () => {
        expect(values).toEqual([1, 2, 3]);
        done();
      }
    });
  });
});

Mocking Observables RxJS

Replace real observables in tests.

import { of, Subject } from 'rxjs';

// Mock service
class MockUserService {
  private users$ = new Subject<User[]>();
  
  getUsers() {
    return this.users$.asObservable();
  }
  
  // Test helper to emit values
  emitUsers(users: User[]) {
    this.users$.next(users);
  }
}

// In test
it('should display users', () => {
  const mockService = new MockUserService();
  const component = new UserListComponent(mockService);
  
  component.ngOnInit();
  
  mockService.emitUsers([
    { id: 1, name: 'Alice' },
    { id: 2, name: 'Bob' }
  ]);
  
  expect(component.users.length).toBe(2);
});

// Using jasmine/jest spies
jest.spyOn(userService, 'getUsers').mockReturnValue(
  of([{ id: 1, name: 'Test User' }])
);

Best Practices

Guidelines for writing clean, efficient RxJS code.

Expert
5 examples

Unsubscribe Properly RxJS

Prevent memory leaks by cleaning up subscriptions.

// ❌ Bad: Subscription leak
class Component {
  ngOnInit() {
    interval(1000).subscribe(val => this.update(val));
    // Never unsubscribed!
  }
}

// ✅ Good: Store and unsubscribe
class Component implements OnDestroy {
  private subscription = new Subscription();

  ngOnInit() {
    this.subscription.add(
      interval(1000).subscribe(val => this.update(val))
    );
  }

  ngOnDestroy() {
    this.subscription.unsubscribe();
  }
}

// ✅ Better: Use takeUntil pattern
class Component implements OnDestroy {
  private destroy$ = new Subject<void>();

  ngOnInit() {
    interval(1000).pipe(
      takeUntil(this.destroy$)
    ).subscribe(val => this.update(val));
  }

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

// ✅ Best: Use async pipe (Angular)
// In template: {{ data$ | async }}

Avoid Nested Subscribes RxJS

Use flattening operators instead of subscribe inside subscribe.

// ❌ Bad: Nested subscribes (callback hell)
getUser().subscribe(user => {
  getPosts(user.id).subscribe(posts => {
    getComments(posts[0].id).subscribe(comments => {
      console.log(comments);
    });
  });
});

// ✅ Good: Flatten with operators
getUser().pipe(
  switchMap(user => getPosts(user.id)),
  switchMap(posts => getComments(posts[0].id))
).subscribe(comments => console.log(comments));

// ✅ With error handling
getUser().pipe(
  switchMap(user => getPosts(user.id).pipe(
    catchError(err => {
      console.error('Failed to get posts');
      return EMPTY;
    })
  )),
  switchMap(posts => posts.length > 0 
    ? getComments(posts[0].id) 
    : of([])
  )
).subscribe(comments => console.log(comments));

Choose the Right Operator RxJS

Selection guide for common scenarios.

// When to use each flattening operator:

// switchMap: Latest only, cancel previous
// - Typeahead search
// - Route parameter changes  
// - HTTP requests where only latest matters
searchInput$.pipe(switchMap(query => search(query)));

// mergeMap: All concurrent
// - Independent parallel operations
// - WebSocket messages
// - Operations where order doesn't matter
clicks$.pipe(mergeMap(e => logClick(e)));

// concatMap: Sequential, maintain order
// - Queue of operations
// - File uploads one by one
// - Ordered writes to server
items$.pipe(concatMap(item => saveItem(item)));

// exhaustMap: Ignore while busy
// - Form submission (prevent double submit)
// - Refresh button (ignore during refresh)
submitBtn$.pipe(exhaustMap(() => saveForm()));

Error Handling Patterns RxJS

Robust error handling strategies.

// Pattern 1: Catch and replace
getData().pipe(
  catchError(err => of(fallbackData))
);

// Pattern 2: Catch, log, and rethrow
getData().pipe(
  catchError(err => {
    logError(err);
    return throwError(() => err);
  })
);

// Pattern 3: Retry with exponential backoff
getData().pipe(
  retry({
    count: 3,
    delay: (err, retryCount) => timer(Math.pow(2, retryCount) * 1000)
  }),
  catchError(err => of(fallbackData))
);

// Pattern 4: Isolate errors in mergeMap
ids$.pipe(
  mergeMap(id => 
    fetchItem(id).pipe(
      catchError(err => {
        console.error(`Failed for :`, err);
        return EMPTY; // Skip this item, continue others
      })
    )
  )
);

// Pattern 5: Global error handler
source$.subscribe({
  next: handleData,
  error: err => {
    showErrorToast(err.message);
    logToServer(err);
  }
});

Performance Tips RxJS

Optimize observable performance.

// 1. Use shareReplay for expensive operations
const data$ = expensiveComputation().pipe(
  shareReplay({ bufferSize: 1, refCount: true })
);

// 2. Debounce/throttle high-frequency events
scroll$.pipe(throttleTime(100));
input$.pipe(debounceTime(300));

// 3. Use distinctUntilChanged to prevent duplicate work
state$.pipe(
  map(s => s.users),
  distinctUntilChanged((a, b) => a.length === b.length)
);

// 4. Limit concurrent operations
urls$.pipe(
  mergeMap(url => fetch(url), 3) // Max 3 concurrent
);

// 5. Take only what you need
infinite$.pipe(take(10));
stream$.pipe(first(x => x > 100));

// 6. Clean up on route changes (Angular)
this.route.params.pipe(
  switchMap(params => this.loadData(params.id)),
  takeUntil(this.destroy$)
);

RxJS Best Practices & Tips

Essential guidelines for writing clean, efficient reactive code.

Memory Management

  • Always unsubscribe from long-lived observables.
  • Use takeUntil with a destroy subject pattern.
  • Prefer async pipe in Angular templates.

Flattening Operators

  • switchMap for cancellation (search, routes).
  • mergeMap for parallel (independent requests).
  • concatMap for order (sequential writes).

Error Handling

  • Use catchError inside inner observables.
  • Implement retry with exponential backoff.
  • Provide fallback values for graceful degradation.

Operator Quick Reference

Operator Category Use Case
mapTransformTransform each value
filterFilterKeep values matching condition
switchMapFlattenCancel previous, use latest
mergeMapFlattenRun all concurrently
concatMapFlattenRun sequentially
debounceTimeFilterWait for pause in emissions
distinctUntilChangedFilterSkip consecutive duplicates
catchErrorErrorHandle errors gracefully
shareReplayMulticastCache and share values
combineLatestCombineLatest from multiple sources
Buy Me A Coffee