Mastering RxJS in Angular: Reactive Programming Patterns

RxJS patterns I copy into every Angular app

Angular ships with RxJS whether you love streams or tolerate them. I am in the second camp on bad days and the first camp when switchMap saves a race condition.

This post is not RxJS from first principles. It is the patterns I copy into real Angular apps — and the leaks I fixed after wondering why the tab felt sluggish.

---

Observables — the mental model

What: A stream of values over time. Subscribe to listen; unsubscribe (or let Angular handle it) to stop.

import { Observable } from "rxjs";

const observable = new Observable<number>((subscriber) => {
  subscriber.next(1);
  subscriber.next(2);
  subscriber.complete();
});

observable.subscribe({
  next: (value) => console.log(value),
  complete: () => console.log("done"),
});

Why Angular cares: HTTP client, router events, forms — observables everywhere.

When I skip RxJS: Simple one-shot HTTP with async pipe and no composition — or Angular signals for local UI state in newer code. RxJS still owns async composition.

---

map, filter, tap — transform pipelines

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

this.userService
  .getUsers()
  .pipe(
    tap((users) => console.log("raw", users.length)),
    map((users) => users.filter((u) => u.active)),
    filter((active) => active.length > 0)
  )
  .subscribe((activeUsers) => {
    this.activeUsers = activeUsers;
  });

tap — side effects (logging), not transformation.

map — transform values.

filter — drop values you do not want downstream.

Red flag: Heavy work inside tap that should be map or a service method.

---

switchMap — cancel stale requests

Perfect for search: user types fast; only the latest query should win.

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

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

  ngOnInit() {
    this.searchSubject
      .pipe(
        debounceTime(300),
        distinctUntilChanged(),
        switchMap((query) => this.userService.searchUsers(query)),
        takeUntil(this.destroy$)
      )
      .subscribe((results) => {
        this.searchResults = results;
      });
  }

  onSearch(query: string) {
    this.searchSubject.next(query);
  }

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

Why switchMap: Cancels the previous inner observable when a new search arrives.

Gotcha: mergeMap when you need every request — switchMap drops in-flight work.

---

combineLatest — multiple sources

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

combineLatest([
  this.userService.getCurrentUser(),
  this.cartService.getCart(),
])
  .pipe(
    map(([user, cart]) => ({
      user,
      canCheckout: cart.items.length > 0 && user.isAuthenticated,
    }))
  )
  .subscribe((data) => {
    this.dashboardData = data;
  });

When: Dashboards where UI depends on two+ live streams.

Gotcha: combineLatest waits for each source to emit at least once — cold HTTP observables need care.

---

Subscriptions — async pipe first

What I prefer:

export class UserListComponent {
  users$ = this.userService.getUsers();
  constructor(private userService: UserService) {}
}

<div *ngFor="let user of users$ | async">{{ user.name }}</div>

Why: Angular unsubscribes for you. Fewer ngOnDestroy bugs.

When I subscribe manually: Imperative side effects (toast, navigation) in the component — and I always pair with cleanup.

---

takeUntilDestroyed (Angular 16+)

import { takeUntilDestroyed } from "@angular/core/rxjs-interop";

export class UserComponent {
  constructor(private userService: UserService) {
    this.userService
      .getUser()
      .pipe(takeUntilDestroyed())
      .subscribe((user) => {
        this.user = user;
      });
  }
}

Why: Less boilerplate than destroy$ Subject pattern.

---

Error handling

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

this.userService
  .getUsers()
  .pipe(
    retry(2),
    catchError(() => of([]))
  )
  .subscribe((users) => {
    this.users = users;
  });

Red flag: Empty catchError that swallows errors silently — log or surface UI.

---

When I reach for signals instead

Angular signals (16+) fit local UI flags — panel open, selected tab — without a Subject. I still use RxJS for HTTP streams, router events, and multi-step async pipelines. Tools overlap; pick the simpler one for the job.

---

Subscription leaks I caused

• Forgot unsubscribe — navigated away, HTTP still updated a destroyed view. async pipe or takeUntilDestroyed fixed most cases.
• Nested subscribe — outer.subscribe(() => inner.subscribe()) instead of switchMap.
• Subjects everywhere — BehaviorSubject for state that belonged in a service with clear API.

---

If you change one thing today

Audit one Angular feature for manual subscriptions. Convert one to async pipe. Fix one search box with debounceTime + switchMap.

RxJS is not the whole app — it is the glue for async flows. Use streams where time and cancellation matter; use simpler tools where they do not.