RxJS and Asynchronous Patterns

This section documents modern, standalone Angular — signals, the built-in @if / @for / @switch control flow, @defer, typed reactive forms, provideHttpClient, functional guards and interceptors, and server-side rendering with hydration — as described by the official documentation at angular.dev, which is the reference these pages are written and verified against.

This content was generated with the assistance of AI and should be verified against angular.dev before being relied on in production. Angular ships a major release roughly every six months and its APIs continue to evolve: the examples here target the current major release; where a consulted source disagrees with the current documentation, the documentation wins and the difference is noted.

This section’s bibliography lists the reference material consulted while preparing these pages.

Angular ships RxJS and uses observables for the HTTP client, router events, reactive form value streams, and @Output events. This page covers the parts you need day to day; for signals, and when to prefer them, see Signals.

From callbacks to observables

A callback delivers one result at some later point; a Promise does the same with chaining and a single resolution. An observable generalises both: it is a lazy, cancellable producer of zero or more values over time.

// callback: one result, no composition
getUser(id, (err, user) => { /* ... */ });

// promise: one result, then-able, not cancellable
const user = await getUser(id);

// observable: many values, lazy (nothing runs until subscribe), cancellable
const sub = interval(1000).subscribe((n) => console.log(n));
sub.unsubscribe(); // stops the producer

  • Lazy — the function passed to new Observable(…​) (or wrapped by of, from, fromEvent) runs only when something subscribes, and re-runs for every subscriber.

  • Cancellable — unsubscribe() tears down timers, listeners, and in-flight HTTP requests.

  • Multi-value — a stream can emit repeatedly (key presses, WebSocket messages, form changes) before it completes, or never complete at all.

For the underlying JavaScript concurrency model (the event loop, microtasks, async/await) see Asynchronous JavaScript. RxJS concepts are introduced at the RxJS overview.

Creating observables

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

of(1, 2, 3);                       // emit each argument, then complete
from(fetch('/api/books'));         // from a Promise (or array, or iterable)
from([10, 20, 30]);
fromEvent(input, 'input');         // from a DOM event target
interval(1000);                    // 0, 1, 2, ... every second (never completes)
timer(2000);                       // emit 0 once after 2s
timer(2000, 1000);                 // first after 2s, then every 1s

A Subject is both an observable and an observer: you call next() on it to push values to every current subscriber. A BehaviorSubject additionally stores the latest value and replays it to new subscribers, which makes it a common building block for small shared state.

import { Subject, BehaviorSubject } from 'rxjs';

const clicks = new Subject<void>();
clicks.subscribe(() => console.log('clicked'));
clicks.next(); // -> "clicked"

@Injectable({ providedIn: 'root' })
export class ThemeService {
  private readonly theme$ = new BehaviorSubject<'light' | 'dark'>('light');
  readonly current$ = this.theme$.asObservable(); // expose read-only
  set(next: 'light' | 'dark') { this.theme$.next(next); }
}

Operators

Operators are pure functions passed to pipe() that transform a stream into a new stream.

Operator Purpose

map, filter, tap

transform each value, drop values, run a side effect without changing the value

switchMap

map to an inner observable, cancelling the previous inner one (searches, autocompletes)

mergeMap

map to an inner observable, run them concurrently (independent writes)

concatMap

map to an inner observable, run them one after another (ordered writes)

debounceTime

wait for a quiet gap before emitting the latest value

distinctUntilChanged

suppress consecutive duplicates

combineLatest

combine the latest value of several streams whenever any of them emits

catchError

handle an error, returning a fallback stream (of(…​), EMPTY) or re-throwing

retry

re-subscribe on error, up to N times or per a delay strategy

takeUntil

complete this stream when a notifier stream emits

A typeahead search combines debounceTime, distinctUntilChanged, and switchMap:

import { Component, inject } from '@angular/core';
import { FormControl, ReactiveFormsModule } from '@angular/forms';
import { HttpClient } from '@angular/common/http';
import { debounceTime, distinctUntilChanged, switchMap, catchError } from 'rxjs/operators';
import { of } from 'rxjs';
import { toSignal } from '@angular/core/rxjs-interop';

@Component({
  selector: 'app-book-search',
  imports: [ReactiveFormsModule],
  template: `
    <input [formControl]="query" placeholder="Search books" />
    @for (book of results(); track book.id) { <li>{{ book.title }}</li> }
  `,
})
export class BookSearch {
  private readonly http = inject(HttpClient);
  readonly query = new FormControl('', { nonNullable: true });

  readonly results = toSignal(
    this.query.valueChanges.pipe(
      debounceTime(300),               // ignore rapid keystrokes
      distinctUntilChanged(),          // skip if the text did not actually change
      switchMap((q) =>                 // cancel the previous request
        this.http.get<Book[]>('/api/books', { params: { q } }).pipe(
          catchError(() => of([])),    // a failed search yields no results, not a broken stream
        ),
      ),
    ),
    { initialValue: [] as Book[] },
  );
}

Operators are documented individually at the RxJS operators guide.

Subscribing and unsubscribing

Every manual subscribe() that does not complete on its own must be torn down, or it leaks (and keeps running after the component is destroyed). There are three good options, roughly in order of preference.

1. The async pipe subscribes in the template and unsubscribes automatically when the view is destroyed. It also marks the component for check on each emission, so it works with OnPush and zoneless change detection.

@Component({
  selector: 'app-books',
  imports: [AsyncPipe],
  template: `
    @if (books$ | async; as books) {
      @for (b of books; track b.id) { <li>{{ b.title }}</li> }
    }
  `,
})
export class Books {
  readonly books$ = inject(BookService).list();
}

See AsyncPipe.

2. takeUntilDestroyed() completes the stream when the current injection context’s DestroyRef fires. Called in a field initializer or constructor it needs no argument; elsewhere pass a captured DestroyRef.

import { Component, inject } from '@angular/core';
import { takeUntilDestroyed } from '@angular/core/rxjs-interop';

@Component({ selector: 'app-ticker', template: '{{ tick }}' })
export class Ticker {
  tick = 0;
  constructor() {
    interval(1000)
      .pipe(takeUntilDestroyed())
      .subscribe(() => (this.tick += 1));
  }
}

3. A manual Subscription collected and unsubscribed in ngOnDestroy — verbose, but explicit:

import { Subscription } from 'rxjs';

export class Ticker implements OnDestroy {
  private readonly sub = new Subscription();
  ngOnInit() {
    this.sub.add(interval(1000).subscribe(/* ... */));
  }
  ngOnDestroy() {
    this.sub.unsubscribe();
  }
}

Bridging to signals

@angular/core/rxjs-interop converts between the two reactive primitives. toSignal() subscribes to an observable and exposes its latest value as a read-only signal (unsubscribing on destroy); toObservable() emits whenever a signal changes.

import { toSignal, toObservable } from '@angular/core/rxjs-interop';

readonly user = toSignal(this.userService.current$, { initialValue: null });
readonly userId = signal(1);
readonly user$ = toObservable(this.userId).pipe(
  switchMap((id) => this.http.get<User>(`/api/users/${id}`)),
);

See the RxJS interop guide.

A pipeline at a glance

flowchart LR A["source$
(valueChanges / fromEvent)"] --> B["debounceTime(300)"] B --> C["distinctUntilChanged()"] C --> D["switchMap(q => http.get(...))
cancels the previous request"] D --> E["catchError(() => of([]))"] E --> F["subscribe() / async pipe / toSignal()"]

Each operator returns a new observable; nothing runs until step F subscribes, and unsubscribe() (or view destruction) propagates back up the chain to cancel timers and the in-flight request.