Published / updated29 Aug 2026

RxJS Basics

Before you start

You need: services (Article 04) and forms (Article 07).

Time: about 50 minutes, plus the practice. RxJS is the part of Angular that confuses people most.

Learning objective

Read and write the RxJS an Angular application actually needs, and avoid the two failure modes it invites: leaked subscriptions and out-of-order responses.

Topics

• What an Observable is
• Creating them
• Subscribing, and unsubscribing
• The operators worth learning
• The flattening operators, and choosing between them
• Subjects
• Error handling
• Combining streams
• Signals and interop

What an Observable is

A Promise produces one value, once, and starts immediately. An Observable produces zero or more values over time, and does nothing until subscribed.

	Promise	Observable
Values	Exactly one	Zero to many
Starts	Immediately	On subscribe
Cancellable	No	Yes
Operators	then, catch	Over a hundred

Lazy is the property that matters. An Observable is a recipe; nothing runs until someone subscribes:

const students$ = this.http.get<Student[]>('/api/students');
// No request has been sent

students$.subscribe(students => console.log(students));
// Now it sends

That catches everyone once — a service method that "does not work" because the caller never subscribed.

The $ suffix on the variable name is the near-universal convention for an Observable.

Creating Observables

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

of(1, 2, 3);                       // emits each, then completes
of([1, 2, 3]);                     // emits ONE array
from([1, 2, 3]);                   // emits each item of the array
from(fetch('/api/students'));      // from a Promise
interval(1000);                    // 0, 1, 2, … every second, forever
timer(2000);                       // one value after 2s
timer(0, 5000);                    // immediately, then every 5s
fromEvent(input, 'input');         // DOM events
EMPTY;                             // completes with no value
throwError(() => new Error('x'));  // errors immediately

of([1,2,3]) versus from([1,2,3]) is a real distinction: of emits the array as a single value, from emits three values. Using the wrong one produces a template that renders nothing.

In Angular, most Observables come from HttpClient, the router, or forms — you rarely create one by hand.

Subscribing and unsubscribing

this.studentService.getStudents().subscribe({
    next: students => this.students.set(students),
    error: err => this.error.set('Could not load students.'),
    complete: () => this.loading.set(false)
});

A subscription that outlives the component is a memory leak. Four ways to prevent it, in order of preference.

1. The async pipe. Subscribes and unsubscribes automatically:

@if (students$ | async; as students) {
    @for (student of students; track student.publicId) { … }
}

2. takeUntilDestroyed.

private readonly destroyRef = inject(DestroyRef);

ngOnInit(): void {
    this.form.controls.className.valueChanges
        .pipe(takeUntilDestroyed(this.destroyRef))
        .subscribe(value => this.loadSections(value));
}

Called in a field initialiser or constructor, takeUntilDestroyed() needs no argument. Called later — in ngOnInit — it needs the DestroyRef.

3. toSignal. Converts to a signal and handles teardown:

readonly students = toSignal(this.studentService.getStudents(), { initialValue: [] });

4. Manual. For completeness; rarely the right answer now:

private readonly subscriptions = new Subscription();

ngOnInit(): void {
    this.subscriptions.add(
        this.service.getStudents().subscribe(students => this.students.set(students)));
}

ngOnDestroy(): void {
    this.subscriptions.unsubscribe();
}

An HttpClient Observable completes after one value, so it unsubscribes itself and does not leak. The leaks come from streams that never complete: valueChanges, paramMap, interval, fromEvent, and any Subject.

The symptom is subtle: navigate away and back ten times, and the handler fires ten times per event. Nothing errors; the application just gets slower.

The operators worth learning

import { map, filter, tap, debounceTime, distinctUntilChanged,
         switchMap, catchError, finalize, take, startWith, shareReplay } from 'rxjs';

this.studentService.getStudents().pipe(
    map(students => students.filter(s => s.status === 'Active')),
    tap(students => console.log(`${students.length} active`)),
    catchError(() => of([]))
).subscribe(students => this.students.set(students));

Operator	Does
map	Transform each value
filter	Drop values that fail a predicate
tap	Side effect, value unchanged — for logging
debounceTime(300)	Wait for a pause before emitting
distinctUntilChanged()	Skip a value identical to the previous
take(1)	Take one value, then complete
startWith(x)	Emit x before anything else
catchError	Handle an error
finalize	Run on complete or error
shareReplay(1)	Share one subscription, replay the last value

tap is for side effects only. Returning a value from it does nothing — that is map.

finalize is the RxJS finally, and the right place to hide a spinner:

this.studentService.search(query).pipe(
    finalize(() => this.loading.set(false))
).subscribe({
    next: results => this.results.set(results),
    error: error => this.errorMessage.set('Search failed.')
});

Hiding it only in next leaves a permanent spinner after an error — the "page froze" report.

The flattening operators

When each value triggers another Observable, one of these decides what happens to the previous one. Choosing wrongly is the most common RxJS bug.

Operator	On a new value	Use for
switchMap	Cancels the previous	Search, navigation, anything superseded
concatMap	Queues, runs in order	Saves that must not reorder
mergeMap	Runs in parallel	Independent work
exhaustMap	Ignores new while one runs	Submit buttons, login

// Search — cancel the previous request
readonly students = toSignal(
    this.searchTerm$.pipe(
        debounceTime(300),
        distinctUntilChanged(),
        switchMap(term => this.studentService.search({ term, page: 1, pageSize: 20 })),
        catchError(() => of({ items: [], totalCount: 0 }))
    ),
    { initialValue: { items: [], totalCount: 0 } });

Typing "Ravi" fires four requests without switchMap, and they can arrive out of order — leaving the list showing results for "Rav". switchMap cancels each superseded request, so only the last result can arrive.

// Submit — ignore repeat clicks while one is in flight
this.saveClicks$.pipe(
    exhaustMap(() => this.studentService.create(this.form.getRawValue()))
).subscribe(student => this.router.navigate(['/students', student.publicId]));

exhaustMap is the RxJS answer to a double-clicked Save creating two students.

// Route parameter — cancel the previous student's request
readonly student = toSignal(
    this.route.paramMap.pipe(
        switchMap(params => this.studentService.getByPublicId(params.get('publicId')!))
    ),
    { initialValue: null });

switchMap for reads, exhaustMap for writes is a good default. mergeMap on a save can reorder writes; switchMap on a save can cancel one half-done.

Subjects

A Subject is both an Observable and an observer — you can push values into it.

import { Subject, BehaviorSubject, ReplaySubject } from 'rxjs';

const clicks$ = new Subject<void>();            // no initial value
const students$ = new BehaviorSubject<Student[]>([]);   // has a current value
const recent$ = new ReplaySubject<string>(3);   // replays the last 3

Type	New subscriber gets
Subject	Only future values
BehaviorSubject	The current value, then future ones
ReplaySubject(n)	The last n values, then future ones

@Injectable({ providedIn: 'root' })
export class StudentStateService {
    private readonly studentsSubject = new BehaviorSubject<Student[]>([]);
    readonly students$ = this.studentsSubject.asObservable();

    setStudents(students: Student[]): void {
        this.studentsSubject.next(students);
    }
}

asObservable() matters. Exposing the Subject lets any component call next(), and finding what changed the state becomes a search through the whole application.

BehaviorSubject is the standard state-sharing primitive in pre-signals Angular. In new code a signal is simpler; in existing code this is what you will find.

Error handling

this.studentService.search(query).pipe(
    catchError((err: HttpErrorResponse) => {
        if (err.status === 404) {
            return of({ items: [], totalCount: 0 });   // recover
        }

        this.error.set('Could not load students.');
        return EMPTY;                                   // give up quietly
    })
).subscribe(result => this.result.set(result));

catchError must return an Observable: of(fallback) to recover, EMPTY to complete silently, or throwError to rethrow.

An error terminates the stream. This is the trap on a long-lived stream:

// Wrong — one failure and search stops working forever
this.searchTerm$.pipe(
    switchMap(term => this.studentService.search(term)),
    catchError(() => of([]))     // catches on the OUTER stream
).subscribe(results => this.results.set(results));

The outer stream errored and completed. No further search term is processed, and the user must reload the page.

// Right — catch inside, so only that request fails
this.searchTerm$.pipe(
    switchMap(term => this.studentService.search(term).pipe(
        catchError(() => of({ items: [], totalCount: 0 }))
    ))
).subscribe(results => this.results.set(results));

Put catchError inside the switchMap. The inner Observable fails; the outer one keeps running.

Retrying transient failures:

import { retry } from 'rxjs';

this.studentService.getStudents().pipe(
    retry({ count: 3, delay: (error, retryCount) => timer(retryCount * 1000) })
).subscribe(students => this.students.set(students));

Retry transient failures only. Retrying a 400 or a 404 fails identically three times and hides the real problem.

Combining streams

import { combineLatest, forkJoin, merge, startWith } from 'rxjs';

// forkJoin — like Promise.all; waits for all to complete
forkJoin({
    students: this.studentService.getStudents(),
    teachers: this.teacherService.getTeachers()
}).subscribe(({ students, teachers }) => {
    this.students.set(students);
    this.teachers.set(teachers);
});

// combineLatest — emits whenever ANY source emits
combineLatest([
    this.searchTerm$.pipe(startWith('')),
    this.className$.pipe(startWith('')),
    this.page$.pipe(startWith(1))
]).pipe(
    debounceTime(300),
    switchMap(([term, className, page]) =>
        this.studentService.search({ term, className, page, pageSize: 20 }))
).subscribe(result => this.result.set(result));

combineLatest is how several filters drive one request. Every source must emit before it emits at all — hence startWith. Omitting it means the request never fires until the user has touched every filter, which reads as a broken page.

forkJoin requires every source to complete. Given a BehaviorSubject, which never completes, it hangs forever with no error.

Signals and interop

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

// Observable → signal
readonly students = toSignal(this.studentService.getStudents(), { initialValue: [] });

// Signal → Observable
readonly searchTerm = signal('');
readonly searchTerm$ = toObservable(this.searchTerm);

readonly results = toSignal(
    toObservable(this.searchTerm).pipe(
        debounceTime(300),
        distinctUntilChanged(),
        switchMap(term => this.studentService.search({ term, page: 1, pageSize: 20 })),
        catchError(() => of({ items: [], totalCount: 0 }))
    ),
    { initialValue: { items: [], totalCount: 0 } });

toSignal subscribes immediately, unsubscribes on destroy, and gives a value the template reads synchronously. initialValue avoids undefined before the first emission.

Signals for state; RxJS for events over time. A current value, a filter, a loading flag — signal. A debounced search, a cancellable request, a websocket — RxJS. The interop functions mean this is not a choice you make once for the whole application.

Angular's httpResource and resource APIs increasingly cover the common fetch-and-render case without RxJS at all.

Debugging

this.searchTerm$.pipe(
    tap(term => console.log('[1] term', term)),
    debounceTime(300),
    tap(term => console.log('[2] debounced', term)),
    switchMap(term => this.studentService.search(term).pipe(
        tap(result => console.log('[3] result', result.totalCount))
    ))
).subscribe();

tap at each stage shows exactly where a value stops flowing.

Symptom	Cause
Nothing happens	Never subscribed
Two HTTP requests for one action	Subscribed twice, or async used twice
Results in the wrong order	mergeMap where switchMap was needed
The stream stops after one error	catchError on the outer stream
Handler fires more each time you visit	Leaked subscription
combineLatest never emits	A source has not emitted — add startWith
forkJoin never emits	A source never completes
Duplicate saves on a double-click	switchMap or mergeMap where exhaustMap belonged

Errors you will hit

What you see	Cause	Fix
Nothing happens	Never subscribed — Observables are lazy	Subscribe, or use the async pipe
The request fires on every keystroke	No debounceTime	Add it with distinctUntilChanged
Results arrive out of order	Used mergeMap for a search	Use switchMap
Search dies after one failure	catchError on the outer stream	Put it inside the switchMap
Memory grows as you navigate	Subscriptions never unsubscribed	async pipe, toSignal, or takeUntilDestroyed
A double-click creates two records	No exhaustMap on the save	Use it

catchError placement decides whether search survives an error. On the outer stream the whole thing completes and never runs again until reload.

Common mistakes

• Never subscribing, so nothing runs
• No teardown on a long-lived stream
• Two | async on the same Observable, causing two requests
• mergeMap on a search, producing out-of-order results
• switchMap on a save, cancelling a half-completed write
• catchError outside switchMap, killing the stream on first failure
• Retrying a 400 or 404
• combineLatest with no startWith
• forkJoin on a BehaviorSubject
• Exposing a Subject instead of asObservable()
• of(array) where from(array) was meant
• Business logic in tap
• Nesting subscribe inside subscribe instead of using a flattening operator

Practice

The course exercise is trace an Observable error.

1. Create a service method returning an Observable and never subscribe. Confirm no request in the Network tab.
2. Subscribe and confirm the request.
3. Use | async twice on the same Observable. Count the requests. Fix with | async; as x.
4. Build a live search with debounceTime and switchMap. Type quickly and confirm one result.
5. Replace switchMap with mergeMap. Throttle to Slow 3G, type quickly, and confirm out-of-order results.
6. Put catchError on the outer stream. Force one failure, then search again. Confirm the stream is dead.
7. Move catchError inside switchMap and confirm recovery.
8. Wire a Save button through exhaustMap. Double-click and confirm one request. Change to switchMap and observe the cancelled first request.
9. Subscribe to valueChanges with no takeUntilDestroyed. Navigate away and back ten times, then type once and count the log lines.
10. Add takeUntilDestroyed and repeat.
11. Combine three filters with combineLatest and no startWith. Confirm nothing loads until every filter is touched.
12. Add startWith and confirm the initial load.
13. Use forkJoin with a BehaviorSubject source and confirm it never emits.
14. Convert a signal to an Observable with toObservable, debounce it, and convert back with toSignal.
15. Add tap logging at three pipeline stages and trace a value through.

Exercises 5, 6 and 9 are the three RxJS bugs that reach production most often.

You can now

• Read and write the RxJS an Angular application needs
• Choose between switchMap, mergeMap, concatMap and exhaustMap
• Put catchError where it does not kill the stream
• Avoid subscription leaks
• Debounce a search correctly

Review questions

1. Why does an Observable do nothing until subscribed?
2. When is switchMap right, and when is exhaustMap?
3. Why must catchError sit inside switchMap on a long-lived stream?
4. Which Observables leak, and which clean up after themselves?

Next: HttpClient and API integration