RxJS Essentials: Most Used Operators and Patterns

Master the most commonly used RxJS operators and patterns for handling async operations, data transformation, and reactive programming in Angular.

By Nitesh Kesarkar5 December 20247 min read

RxJS
Reactive Programming
TypeScript
Angular
Observables

RxJS Essentials: Most Used Operators and Patterns

RxJS is the foundation of reactive programming in Angular. Understanding the most commonly used operators is crucial for building efficient Angular applications. After working extensively with RxJS in production applications, here are the operators and patterns you'll use most frequently.

Core Concepts

RxJS revolves around Observables, which represent streams of values over time:

TSX

import { Observable } from 'rxjs';

// Create an observable
const numbers$ = new Observable<number>(observer => {
  observer.next(1);
  observer.next(2);
  observer.next(3);
  observer.complete();
});

// Subscribe to values
numbers$.subscribe({
  next: value => console.log(value),
  complete: () => console.log('Done'),
});

Most Used Operators

map - Transform Values

Transform each emitted value:

TSX

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

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

numbers$.pipe(
  map(x => x * 2)
).subscribe(value => console.log(value)); // 2, 4, 6

// Transform objects
interface User {
  id: number;
  name: string;
}

const users$ = of<User[]>([]);

users$.pipe(
  map(users => users.map(user => user.name))
).subscribe(names => console.log(names));

filter - Filter Values

Filter emitted values based on condition:

TSX

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

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

numbers$.pipe(
  filter(x => x % 2 === 0)
).subscribe(value => console.log(value)); // 2, 4

// Filter objects
interface Product {
  id: string;
  price: number;
  inStock: boolean;
}

products$.pipe(
  filter(product => product.inStock && product.price < 100)
).subscribe(filteredProducts => {
  // Handle filtered products
});

tap - Side Effects

Perform side effects without modifying the stream:

TSX

import { tap } from 'rxjs/operators';

users$.pipe(
  tap(users => console.log('Received users:', users)),
  tap(users => this.updateCache(users)),
  map(users => users.filter(u => u.active))
).subscribe(activeUsers => {
  // Process active users
});

switchMap - Switch to New Observable

Cancel previous inner observable when new value arrives:

TSX

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

// Perfect for search inputs
fromEvent(searchInput, 'input').pipe(
  map(event => (event.target as HTMLInputElement).value),
  switchMap(searchTerm => 
    this.searchService.search(searchTerm)
  )
).subscribe(results => {
  this.results = results;
});

mergeMap (flatMap) - Flatten Observables

Flatten inner observables:

TSX

import { mergeMap } from 'rxjs/operators';

// Fetch details for each user
users$.pipe(
  mergeMap(user => this.userService.getUserDetails(user.id))
).subscribe(userDetails => {
  // Handle each user's details
});

concatMap - Sequential Execution

Execute observables sequentially:

TSX

import { concatMap } from 'rxjs/operators';

// Process items one at a time
items$.pipe(
  concatMap(item => this.processItem(item))
).subscribe(result => {
  // Results arrive in order
});

debounceTime - Debounce Values

Emit value after specified time has passed:

TSX

import { debounceTime } from 'rxjs/operators';

// Debounce search input
searchInput$.pipe(
  debounceTime(300)
).subscribe(searchTerm => {
  // Only fires 300ms after last input
  this.performSearch(searchTerm);
});

distinctUntilChanged - Skip Duplicates

Skip values that are the same as previous:

TSX

import { distinctUntilChanged } from 'rxjs/operators';

filter$.pipe(
  distinctUntilChanged()
).subscribe(filter => {
  // Only fires when filter actually changes
  this.applyFilter(filter);
});

catchError - Error Handling

Handle errors gracefully:

TSX

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

this.userService.getUsers().pipe(
  catchError(error => {
    console.error('Error fetching users:', error);
    return of([]); // Return empty array on error
  })
).subscribe(users => {
  this.users = users;
});

takeUntil - Complete on Signal

Complete observable when another emits:

TSX

import { takeUntil } from 'rxjs/operators';
import { Subject } from 'rxjs';

export class Component implements OnInit, OnDestroy {
  private destroy$ = new Subject<void>();

  ngOnInit(): void {
    this.userService.getUsers().pipe(
      takeUntil(this.destroy$)
    ).subscribe(users => {
      this.users = users;
    });
  }

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

Share observable and replay last value:

TSX

import { shareReplay } from 'rxjs/operators';

// Cache and share HTTP request
const users$ = this.http.get<User[]>('/api/users').pipe(
  shareReplay(1) // Cache last value, share subscription
);

// Multiple subscribers share the same request
users$.subscribe(users => console.log('Subscriber 1:', users));
users$.subscribe(users => console.log('Subscriber 2:', users));

combineLatest - Combine Latest Values

Combine latest values from multiple observables:

TSX

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

combineLatest([
  this.filter$,
  this.sort$,
  this.users$
]).pipe(
  map(([filter, sort, users]) => {
    // Combine all values
    return this.applyFilterAndSort(users, filter, sort);
  })
).subscribe(filteredUsers => {
  this.displayUsers = filteredUsers;
});

forkJoin - Wait for All

Wait for all observables to complete:

TSX

import { forkJoin } from 'rxjs';

forkJoin({
  users: this.userService.getUsers(),
  products: this.productService.getProducts(),
  orders: this.orderService.getOrders()
}).subscribe(({ users, products, orders }) => {
  // All requests completed
  this.initializeDashboard(users, products, orders);
});

Common Patterns

Search Pattern

TSX

import { Subject } from 'rxjs';
import { debounceTime, distinctUntilChanged, switchMap } from 'rxjs/operators';

export class SearchComponent {
  private searchTerm$ = new Subject<string>();
  results$ = this.searchTerm$.pipe(
    debounceTime(300),
    distinctUntilChanged(),
    switchMap(term => this.searchService.search(term))
  );

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

Polling Pattern

TSX

import { interval } from 'rxjs';
import { switchMap } from 'rxjs/operators';

// Poll every 5 seconds
interval(5000).pipe(
  switchMap(() => this.dataService.getData())
).subscribe(data => {
  this.updateData(data);
});

Retry Pattern

TSX

import { retry, retryWhen, delay, take } from 'rxjs/operators';

this.apiService.getData().pipe(
  retry(3) // Retry 3 times
).subscribe(data => {
  // Handle data
});

// Retry with delay
this.apiService.getData().pipe(
  retryWhen(errors =>
    errors.pipe(
      delay(1000),
      take(3)
    )
  )
).subscribe(data => {
  // Handle data
});

Loading State Pattern

TSX

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

export class DataService {
  private loading$ = new BehaviorSubject<boolean>(false);

  get loading(): Observable<boolean> {
    return this.loading$.asObservable();
  }

  getData(): Observable<Data[]> {
    this.loading$.next(true);
    return this.http.get<Data[]>('/api/data').pipe(
      catchError(error => {
        console.error(error);
        return of([]);
      }),
      finalize(() => this.loading$.next(false))
    );
  }
}

Real-World Example

Complete example from production:

TSX

import { Component, OnInit, OnDestroy } from '@angular/core';
import { Subject, combineLatest } from 'rxjs';
import { debounceTime, distinctUntilChanged, switchMap, takeUntil, map } from 'rxjs/operators';
import { PlayerService } from './player.service';

interface Player {
  id: string;
  name: string;
  position: string;
  active: boolean;
}

interface Filters {
  search: string;
  position: string;
}

@Component({
  selector: 'app-player-list',
  template: `
    <input 
      (input)="searchTerm$.next($event.target.value)"
      placeholder="Search"
    />
    <select (change)="positionFilter$.next($event.target.value)">
      <option value="">All</option>
      <option value="forward">Forward</option>
    </select>
    <div *ngFor="let player of filteredPlayers$ | async">
      {{ player.name }}
    </div>
  `,
})
export class PlayerListComponent implements OnInit, OnDestroy {
  private destroy$ = new Subject<void>();
  searchTerm$ = new Subject<string>();
  positionFilter$ = new Subject<string>();

  players$ = this.playerService.getPlayers();

  filteredPlayers$ = combineLatest([
    this.players$,
    this.searchTerm$.pipe(
      debounceTime(300),
      distinctUntilChanged(),
      map(term => term.toLowerCase())
    ),
    this.positionFilter$.pipe(distinctUntilChanged())
  ]).pipe(
    map(([players, searchTerm, position]) => {
      return players.filter(player => {
        const matchesSearch = player.name.toLowerCase().includes(searchTerm);
        const matchesPosition = !position || player.position === position;
        return matchesSearch && matchesPosition;
      });
    }),
    takeUntil(this.destroy$)
  );

  ngOnInit(): void {
    this.searchTerm$.next('');
    this.positionFilter$.next('');
  }

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

Best Practices

Use async pipe: Automatic subscription management
Unsubscribe properly: Use takeUntil or async pipe
Combine operators: Chain operators for complex logic
Handle errors: Always use catchError
Debounce inputs: Prevent excessive API calls
Use shareReplay: Cache expensive operations
Choose right operator: switchMap vs mergeMap vs concatMap
Avoid nested subscriptions: Use operators instead

Operator Cheat Sheet

Transformation: map, pluck, scan
Filtering: filter, take, skip, distinctUntilChanged
Combination: merge, concat, combineLatest, forkJoin
Error handling: catchError, retry, retryWhen
Utility: tap, delay, debounceTime, throttleTime
Flattening: switchMap, mergeMap, concatMap, exhaustMap

RxJS is powerful but can be complex. Start with the most common operators (map, filter, switchMap, debounceTime, catchError) and gradually expand your knowledge. The key is understanding when to use each operator for your specific use case.

Keep reading

25 December 2024

NgRx State Management: Enterprise-Ready Angular Architecture

Master NgRx for scalable Angular applications. Learn actions, reducers, effects, and selectors to build maintainable state management solutions.

20 December 2024

Angular Signals: The Future of Reactive Programming

Explore Angular Signals, a new reactive primitive that simplifies state management and improves performance with fine-grained reactivity.

10 December 2024

Angular Performance Optimization: A Complete Guide

Learn advanced techniques to optimize Angular applications, from change detection to lazy loading and bundle optimization.