I have been writing Angular professionally since 2017. I have seen teams drown in operator chains nobody could debug, and I have seen teams underuse RxJS and end up with race conditions that took weeks to track down. The gap between both extremes is smaller than people think. You need about ten operators, understood deeply.
switchMap: Cancel and Replace
switchMap is the operator I reach for first when user input drives an HTTP call. When a new value arrives from the source, switchMap unsubscribes from the previous inner observable and starts a new one. For search-as-you-type, this means the in-flight request from the previous keystroke gets cancelled and replaced.
I used this extensively on the Wonderlabs credit card platform. The card and transaction filters refetched on every change. Without switchMap, responses would arrive out of order and overwrite newer results with older ones. That is a subtle, hard-to-reproduce bug in production.
@Component({ selector: 'app-product-search', standalone: true, template: '' })
export class ProductSearchComponent {
private productService = inject(ProductService);
searchControl = new FormControl('');
results: Product[] = [];
// In the constructor, so bare takeUntilDestroyed() has an injection context
constructor() {
this.searchControl.valueChanges.pipe(
debounceTime(350),
distinctUntilChanged(),
switchMap(query => this.productService.search(query ?? '')),
takeUntilDestroyed()
).subscribe(results => this.results = results);
}
}exhaustMap: Ignore Until Done
exhaustMap is the underrated one. While an inner observable is active, it ignores all new source emissions. The canonical use case is a form submit button. Put exhaustMap behind the submit button of a loan application form: if the user double-clicks or clicks again during the API call, the second click is silently ignored. The request is not duplicated.
@Component({ selector: 'app-application-form', standalone: true, template: '' })
export class ApplicationFormComponent {
private lendingApi = inject(LendingApiService);
private router = inject(Router);
submitClicks$ = new Subject<void>();
constructor() {
this.submitClicks$.pipe(
exhaustMap(() => this.lendingApi.submitApplication(this.form.value).pipe(
catchError(err => {
this.handleError(err);
return EMPTY;
})
)),
takeUntilDestroyed()
).subscribe(confirmation => {
this.router.navigate(['/confirmation', confirmation.id]);
});
}
}Note the catchError inside the exhaustMap's projection function. If the error propagates to the outer pipe, exhaustMap terminates and future clicks stop working. Always catch inside the inner observable.
mergeMap: Run Everything in Parallel
Where switchMap cancels and exhaustMap ignores, mergeMap subscribes to every inner observable concurrently. Every source emission produces an active subscription. Use this when order does not matter and you want all emissions handled.
// Upload multiple documents in parallel for a loan application
const uploads$ = from(selectedDocuments).pipe(
mergeMap(doc => this.documentService.upload(doc).pipe(
catchError(() => of({ id: doc.id, status: 'failed' }))
))
);
uploads$.subscribe(result => this.updateUploadStatus(result));concatMap: Queue and Preserve Order
concatMap is mergeMap with a queue. It subscribes to one inner observable at a time and waits for it to complete before starting the next. Use it when order is part of the contract: sequential uploads that the server stitches back together, audit events that must land in the order the user produced them, a wizard that saves each step before the next one starts. The cost is back pressure. If the source emits faster than the inner observable completes, the queue grows without a bound and nothing tells you. If you cannot argue that the source is slower than the work, use mergeMap with a concurrency limit instead.
@Component({ selector: 'app-chunked-upload', standalone: true, template: '' })
export class ChunkedUploadComponent {
private uploadService = inject(UploadService);
private destroyRef = inject(DestroyRef);
send(chunks: Chunk[]): void {
// Each chunk must land before the next one starts
from(chunks).pipe(
concatMap(chunk => this.uploadService.sendChunk(chunk)),
takeUntilDestroyed(this.destroyRef)
).subscribe(ack => this.markChunkStored(ack.index));
}
sendUnordered(chunks: Chunk[]): void {
// mergeMap with a concurrency cap when order does not matter but load does
from(chunks).pipe(
mergeMap(chunk => this.uploadService.sendChunk(chunk), 4),
takeUntilDestroyed(this.destroyRef)
).subscribe();
}
}combineLatest: React to Multiple Streams at Once
combineLatest emits whenever any of its sources emit, combining the latest value from each. I used this on the real-time investment dashboards at GFT. A dashboard filtered by date range, region, and product type needs to refetch whenever any filter changes.
@Component({ selector: 'app-report-dashboard', standalone: true, template: '' })
export class ReportDashboardComponent {
private reportService = inject(ReportService);
reportData: ReportRow[] = [];
constructor() {
combineLatest([
this.dateRangeControl.valueChanges.pipe(startWith(this.dateRangeControl.value)),
this.regionControl.valueChanges.pipe(startWith(this.regionControl.value)),
this.productControl.valueChanges.pipe(startWith(this.productControl.value))
]).pipe(
debounceTime(200),
switchMap(([dateRange, region, product]) =>
this.reportService.fetch({ dateRange, region, product })
),
takeUntilDestroyed()
).subscribe(data => this.reportData = data);
}
}The startWith on each control is important. combineLatest will not emit until all sources have emitted at least once. Without startWith, the dashboard waits for the user to interact with all three filters before loading anything.
forkJoin: Wait for All to Complete
forkJoin is Promise.all for observables. It waits for all sources to complete and emits their final values together. I use it in route resolvers and initialization logic.
@Component({ selector: 'app-lending-page', standalone: true, template: '' })
export class LendingPageComponent implements OnInit {
private destroyRef = inject(DestroyRef);
ngOnInit(): void {
// Page init: load user, permissions, and feature config before rendering.
// Outside the constructor, takeUntilDestroyed needs the DestroyRef passed in.
forkJoin({
user: this.userService.getCurrentUser(),
permissions: this.authService.getPermissions(),
config: this.featureService.getLendingConfig()
}).pipe(
takeUntilDestroyed(this.destroyRef)
).subscribe(({ user, permissions, config }) => {
this.initializePage(user, permissions, config);
});
}
}One gotcha: if any source in forkJoin errors, the whole thing errors. Wrap individual observables with catchError if you need partial results on failure.
debounceTime and distinctUntilChanged: Always Together
These two are almost never used apart. debounceTime waits for a quiet period. distinctUntilChanged drops the emission if the value has not changed. Together they prevent the most common API abuse pattern: hammering a search endpoint on every keystroke with the same value repeated.
@Component({ selector: 'app-filter-panel', standalone: true, template: '' })
export class FilterPanelComponent {
private api = inject(SearchApiService);
filterControl = new FormControl('');
results: SearchResult[] = [];
constructor() {
this.filterControl.valueChanges.pipe(
debounceTime(400),
distinctUntilChanged(),
switchMap(term => this.api.filter(term ?? '')),
takeUntilDestroyed()
).subscribe(results => this.results = results);
}
}Subscription Management: takeUntilDestroyed
The old pattern was a Subject, takeUntil, and ngOnDestroy. Angular 16 introduced takeUntilDestroyed from @angular/core/rxjs-interop, which hooks into the component's destroy lifecycle automatically. On any codebase running Angular 16 or later, this is the pattern I use.
import { takeUntilDestroyed } from '@angular/core/rxjs-interop';
import { inject, DestroyRef } from '@angular/core';
@Component({ standalone: true, template: '' })
export class ExampleComponent {
// Call outside of a constructor needs explicit DestroyRef
private destroyRef = inject(DestroyRef);
ngOnInit() {
this.someObservable$.pipe(
takeUntilDestroyed(this.destroyRef)
).subscribe();
}
// Or in constructor, no argument needed:
constructor() {
this.otherObservable$.pipe(
takeUntilDestroyed() // Infers context automatically
).subscribe();
}
}The One Operator People Forget: catchError Placement
This is not an operator tip but an operator mistake. When you put catchError at the outer pipe level and an error occurs, the entire stream terminates. If that stream drives a UI element, it stops responding forever. Always put catchError inside the inner observable (inside switchMap, exhaustMap, etc.) and return EMPTY or a fallback value. The outer stream stays alive.
// Wrong: error kills the entire search stream
this.searchControl.valueChanges.pipe(
switchMap(q => this.api.search(q)),
catchError(err => { // Stream terminates here. No more searches work.
this.showError(err);
return of([]);
})
);
// Correct: error stays inside the inner observable
this.searchControl.valueChanges.pipe(
switchMap(q => this.api.search(q).pipe(
catchError(err => {
this.showError(err);
return of([]); // inner observable completes cleanly
})
))
);The goal is not knowing every RxJS operator. It is knowing a small set deeply enough that you reach for the right one without thinking.
switchMap for cancellable requests. exhaustMap for form submissions. mergeMap for parallel independent operations. concatMap when order is part of the contract. forkJoin for page-level data loading. combineLatest for multi-filter dashboards. debounceTime plus distinctUntilChanged on every user input stream. takeUntilDestroyed on every subscription. catchError inside the inner observable. That is the full list.
References
- RxJS operator documentation. https://rxjs.dev/api
- switchMap. https://rxjs.dev/api/operators/switchMap
- exhaustMap. https://rxjs.dev/api/operators/exhaustMap
- concatMap. https://rxjs.dev/api/operators/concatMap
- takeUntilDestroyed. https://angular.dev/api/core/rxjs-interop/takeUntilDestroyed