RxJS Mapping and Combination Operators

Lesson content

RxJS Mapping and Combination Operators RxJS operators help transform Observable values, manage asynchronous tasks, and combine multiple data streams. Simple Definition Mapping Operators = Transform Values or Map Values to Observables Combination Operators = Combine Multiple Observable Streams Main Operator Categories Category Operators Value mapping map Higher-order mapping switchMap , mergeMap , concatMap , exhaustMap Continuous combination combineLatest , withLatestFrom Completion-based combination forkJoin Pair-by-position combination zip Concurrent stream combination merge , concat 1. map The map operator transforms every emitted value into another value. import { map, of } from 'rxjs'; of( 10, 20, 30 ).pipe( map(value => value * 2 ) ).subscribe( result => console.log(result) ); Output 20 40 60 Flow 10 → map → 20 20 → map → 40 30 → map → 60 Object Transformation this.userService .getUsers() .pipe( map(users => users.map(user => ({ id: user.id, displayName: user.name.toUpperCase() })) ) ); Use map When Transforming values Selecting object properties Calculating derived values Changing an API response shape Important Use map when the callback returns a normal value. map(user => user.name) Do not use it to handle nested HTTP Observables. Higher-Order Mapping Operators A higher-order mapping operator maps each source value to another Observable and then manages those inner Observables. Source Observable ↓ Value ↓ Map to Inner Observable ↓ Flatten Inner Observable ↓ Final Result 2. switchMap The switchMap operator switches to the newest inner Observable. When a new source value arrives, it unsubscribes from the previous inner Observable. searchControl.valueChanges .pipe( debounceTime(300), distinctUntilChanged(), switchMap(searchText => this.http.get<Product[]>( '/api/products', { params: { search: searchText } } ) ) ); Flow Search "a" ↓ Request A Starts Search "an" ↓ Request A Cancelled Request B Starts Search "ang" ↓ Request B Cancelled Request C Starts Best Use Cases Search autocomplete Route-parameter API calls Reloading data when filters change Operations where only the latest result matters Avoid switchMap When Do not use it when every operation must complete, such as saving multiple independent records. 3. mergeMap The mergeMap operator runs inner Observables concurrently. It does not cancel previous inner Observables when new values arrive. from(fileList) .pipe( mergeMap(file => this.uploadService .upload(file) ) ) .subscribe(); Flow Value A → Request A ──────┐ Value B → Request B ──┐ │ Value C → Request C ─────┐│ ↓↓↓ Results as Completed Best Use Cases Parallel file uploads Independent API operations Processing tasks concurrently Operations where every result matters Concurrency Limit from(files).pipe( mergeMap( file => uploadFile(file), 3 ) ); This example processes at most three uploads concurrently. 4. concatMap The concatMap operator runs inner Observables one at a time and preserves their order. The next operation waits until the previous one completes. from(commands) .pipe( concatMap(command => this.api.execute( command ) ) ) .subscribe(); Flow Value A ↓ Request A Completes ↓ Value B ↓ Request B Completes ↓ Value C Best Use Cases Ordered saves Sequential API requests Queue processing Operations that must not overlap Important If an inner Observable takes a long time to complete, later values remain queued. 5. exhaustMap The exhaustMap operator ignores new source values while the current inner Observable is active. saveClicks$ .pipe( exhaustMap(() => this.userService .saveUser( this.form.getRawValue() ) ) ) .subscribe(); Flow Click 1 ↓ Save Request Starts Click 2 → Ignored Click 3 → Ignored Save Completes Click 4 ↓ New Save Request Starts Best Use Cases Preventing repeated form submission Login button clicks Payment button protection Ignoring repeated actions until completion Mapping Operator Comparison Operator Previous Request New Values Common Use switchMap Cancels it Uses latest value Search mergeMap Keeps it running Runs concurrently Parallel tasks concatMap Waits for it Queues new values Ordered saves exhaustMap Keeps it running Ignores new values Prevent double submit Easy Memory Formula switchMap = Latest Wins mergeMap = All Run Together concatMap = One After Another exhaustMap = First Wins Until Complete 6. combineLatest The combineLatest function combines the latest value from each source Observable. After every source has emitted at least once, it emits whenever any source emits a new value. import { combineLatest } from 'rxjs'; const filters$ = combineLatest([ category$, searchText$, sortOrder$ ]); filters$.subscribe( ([ category, searchText, sortOrder ]) => { console.log( category, searchText, sortOrder ); } ); Flow Category$ → Electronics Search$ → Keyboard SortOrder$ → Price combineLatest emits: [ 'Electronics', 'Keyboard', 'Price' ] When One Source Changes Search$ → Mouse New output: [ 'Electronics', 'Mouse', 'Price' ] Best Use Cases Combining filters Combining form controls Creating derived application state Loading data based on several parameters Important combineLatest does not emit until every source has emitted at least one value. combineLatest([ category$, search$ ]) Use initial values when necessary. combineLatest([ category$.pipe( startWith('all') ), search$.pipe( startWith('') ) ]) combineLatest with switchMap products$ = combineLatest([ this.category$, this.searchText$, this.page$ ]).pipe( switchMap( ([ category, search, page ]) => this.productService .getProducts({ category, search, page }) ) ); Filter Changes ↓ combineLatest Creates Parameters ↓ switchMap Cancels Old Request ↓ Latest Products Loaded 7. withLatestFrom The withLatestFrom operator combines the source value with the latest values from other Observables. It emits only when the main source Observable emits. saveClick$ .pipe( withLatestFrom( currentUser$, formValue$ ) ) .subscribe( ([ click, user, formValue ]) => { console.log( user, formValue ); } ); Flow Current User Changes → No Output Form Value Changes → No Output Save Button Emits ↓ Output: [ saveEvent, latestUser, latestFormValue ] Best Use Cases Combine a button click with current state Add authentication data to an action Use the latest store value when an event occurs Submit the latest form data combineLatest vs withLatestFrom combineLatest withLatestFrom All sources can trigger output Only the main source triggers output Used for reactive derived state Used for event plus current state Common for filters Common for button clicks Creation function Pipeable operator 8. forkJoin The forkJoin function waits for all source Observables to complete and then emits their final values once. import { forkJoin } from 'rxjs'; forkJoin({ user: this.http.get<User>( '/api/user/10' ), orders: this.http.get<Order[]>( '/api/user/10/orders' ), settings: this.http.get<Settings>( '/api/settings' ) }).subscribe(result => { console.log( result.user, result.orders, result.settings ); }); Flow User Request ──────── Complete Orders Request ── Complete Settings Request ─────── Complete ↓ forkJoin Emits Once Best Use Cases Parallel HTTP requests Page initialization requiring several APIs Waiting for all independent tasks Combining final results Important Warning If any source never completes, forkJoin never emits. Avoid: forkJoin([ interval(1000), httpRequest$ ]) The interval does not complete. Error Behavior When one source errors, the combined forkJoin also errors unless the error is handled inside that source. forkJoin({ user: userRequest$.pipe( catchError(() => of(null) ) ), orders: ordersRequest$ }); 9. zip The zip function combines values by their matching emission position. import { of, zip } from 'rxjs'; zip( of( 'Ajay', 'Sneha', 'Rahul' ), of( 10, 20, 30 ) ).subscribe( result => console.log(result) ); Output ['Ajay', 10] ['Sneha', 20] ['Rahul', 30] Flow Stream A: A1 → A2 → A3 Stream B: B1 → B2 → B3 zip Output: [A1, B1] [A2, B2] [A3, B3] Best Use Cases Pairing values by position Combining synchronized sequences Matching related emissions one by one 10. merge The merge function subscribes to multiple Observables and emits values from all of them as they arrive. import { merge } from 'rxjs'; merge( buttonClick$, keyboardEvent$, refreshEvent$ ).subscribe(event => { console.log(event); }); Flow Stream A ──A────A──── Stream B ─────B──B─── Stream C ─C────────C─ merge ─C A B A B C─ Best Use Cases Combining several event sources Listening to mouse and keyboard actions Combining manual and automatic refresh events 11. concat The concat function subscribes to Observables one after another. The next Observable starts only after the previous Observable completes. import { concat, of } from 'rxjs'; concat( of( 'First 1', 'First 2' ), of( 'Second 1', 'Second 2' ) ).subscribe( value => console.log(value) ); Output First 1 First 2 Second 1 Second 2 Combination Operator Comparison Operator When It Emits Common Use combineLatest Whenever any source changes after all have emitted Reactive filters and state withLatestFrom When the main source emits Event plus latest state forkJoin Once after all sources complete Parallel HTTP calls zip When each source has the next matching value Pair values by position merge Whenever any source emits Multiple event sources concat From one stream at a time Sequential streams Practical Angular Search Example import { Component, inject } from '@angular/core'; import { FormControl, ReactiveFormsModule } from '@angular/forms'; import { AsyncPipe } from '@angular/common'; import { debounceTime, distinctUntilChanged, startWith, switchMap } from 'rxjs'; @Component({ selector: 'app-product-search', imports: [ ReactiveFormsModule, AsyncPipe ], template: ` <input [formControl]="searchControl" placeholder="Search products" /> @if ( products$ | async; as products ) { @for ( product of products; track product.id ) { <p> {{ product.name }} </p> } } ` }) export class ProductSearchComponent { private productService = inject(ProductService); searchControl = new FormControl( '', { nonNullable: true } ); products$ = this.searchControl .valueChanges .pipe( startWith(''), debounceTime(300), distinctUntilChanged(), switchMap(search => this.productService .searchProducts( search ) ) ); } Angular Signals with RxJS Angular provides utilities for converting between signals and RxJS Observables. Signal to Observable import { signal } from '@angular/core'; import { toObservable } from '@angular/core/rxjs-interop'; import { switchMap } from 'rxjs'; searchText = signal(''); searchText$ = toObservable( this.searchText ); products$ = this.searchText$ .pipe( switchMap(search => this.productService .searchProducts( search ) ) ); Observable to Signal import { toSignal } from '@angular/core/rxjs-interop'; products = toSignal( this.products$, { initialValue: [] } ); Error Placement in Mapping Operators The position of catchError changes whether only one inner request fails or the entire outer stream ends. Catch Inside switchMap search$.pipe( switchMap(search => this.api .search(search) .pipe( catchError(() => of([]) ) ) ) ); A failed request returns an empty list, but future search values can still trigger new requests. Catch Outside switchMap search$.pipe( switchMap(search => this.api.search(search) ), catchError(() => of([]) ) ); The outer stream handles the error and may complete after emitting the fallback value. Common Mistakes 1. Using map for an HTTP Observable Avoid: userId$.pipe( map(id => this.http.get( `/api/users/${id}` ) ) ); This creates an Observable of Observables. Prefer: userId$.pipe( switchMap(id => this.http.get( `/api/users/${id}` ) ) ); 2. Using switchMap for Required Saves A new emission can cancel the previous inner operation. Use concatMap for ordered operations or mergeMap for independent concurrent operations. 3. Using mergeMap Without a Concurrency Limit A fast source can create too many simultaneous requests. 4. Using concatMap with a Never-Completing Inner Stream Later values remain queued forever when the current inner Observable never completes. 5. Using forkJoin with Long-Lived Streams Use forkJoin only when every source completes. 6. Expecting combineLatest to Emit Immediately Every source must emit at least once before the first combined output. 7. Confusing combineLatest and withLatestFrom Use combineLatest when all sources can trigger updates. Use withLatestFrom when one main source controls emission. 8. Nested Subscriptions Avoid: userId$.subscribe(id => { this.http .get(`/api/users/${id}`) .subscribe(user => { console.log(user); }); }); Prefer: userId$ .pipe( switchMap(id => this.http.get( `/api/users/${id}` ) ) ) .subscribe(user => { console.log(user); }); Best Practices Use map for normal value transformations Use switchMap when only the latest result matters Use mergeMap for independent concurrent operations Use concatMap for ordered sequential operations Use exhaustMap to ignore repeated actions while one is active Use combineLatest for continuously derived state Use withLatestFrom for an event combined with current state Use forkJoin for finite parallel operations Avoid nested subscriptions Place error handling at the correct stream level Limit concurrency when processing many tasks Ensure completion-based operators receive completing streams Interview Questions What is the difference between map and switchMap? map transforms a value into another normal value. switchMap maps a value to an Observable and switches to the newest inner Observable. What is the difference between switchMap and mergeMap? switchMap cancels the previous inner subscription when a new value arrives. mergeMap keeps all inner subscriptions running concurrently. When should concatMap be used? Use it when asynchronous operations must run one at a time and preserve their order. When should exhaustMap be used? Use it when repeated source emissions should be ignored until the current operation completes. What is combineLatest? It emits the latest values from all sources whenever any source emits, after every source has emitted at least once. What is withLatestFrom? It combines a source emission with the latest values from other Observables, but only the main source triggers an output. What is forkJoin? It waits for all source Observables to complete and emits their final values once. What happens when a forkJoin source never completes? The combined Observable never emits its result. Why should nested subscriptions be avoided? Nested subscriptions make cancellation, errors, sequencing, and cleanup harder to manage. Higher-order mapping operators provide a clearer stream. Key Point Choose an RxJS operator according to how new emissions should affect active asynchronous operations and when combined streams should produce output. Easy Formula map = Value → Value switchMap = Latest Wins mergeMap = Run All Concurrently concatMap = Queue in Order exhaustMap = Ignore New Until Complete combineLatest = Emit When Any Source Changes withLatestFrom = Main Source + Latest State forkJoin = Wait for All to Complete zip = Match by Position merge = Emit from Every Stream concat = Streams One After Another