switchMap, mergeMap, and combineLatest

Flattening operators handle Observables that emit Observables (higher-order Observables). They subscribe to inner Observables and flatten the result.

switchMap — Cancel Previous

Cancels the previous inner Observable when a new value arrives. Ideal for type-ahead search.

import { switchMap } from 'rxjs/operators';

this.searchInput.valueChanges.pipe(
  debounceTime(300),
  switchMap(query => this.searchService.search(query))
).subscribe(results => {
  this.results = results;
});

When the user types a new character, the previous search request is cancelled and a new one starts.

mergeMap — Concurrent

Runs all inner Observables concurrently without waiting. Ideal for parallel requests.

import { mergeMap } from 'rxjs/operators';

this.userIds$.pipe(
  mergeMap(ids => this.http.get<User[]>(`/api/users?ids=${ids.join(',')}`))
).subscribe(users => {
  this.users = users;
});

concatMap — Sequential

Runs inner Observables one at a time, in order. Ideal for dependent requests.

import { concatMap } from 'rxjs/operators';

this.saveRequests$.pipe(
  concatMap(request => this.http.post('/api/save', request))
).subscribe(result => {
  console.log('Saved:', result);
});

Flattening Operator Comparison

Practical Example: Search with switchMap

@Component({
  selector: 'app-search',
  template: `
    <input [formControl]="searchControl" placeholder="Search...">
    <p *ngFor="let result of results">{{ result.name }}</p>
  `,
})
export class SearchComponent implements OnInit {
  searchControl = new FormControl('');
  results: SearchResult[] = [];

  ngOnInit(): void {
    this.searchControl.valueChanges.pipe(
      debounceTime(300),
      distinctUntilChanged(),
      switchMap(query => this.searchService.search(query)),
      catchError(() => of([]))
    ).subscribe(results => {
      this.results = results;
    });
  }
}

Joining operators combine multiple Observables into one.

combineLatest — Latest from All

Emits whenever any source emits, using the latest value from each source.

import { combineLatest } from 'rxjs';

const price$ = this.getPrice();
const quantity$ = this.getQuantity();

combineLatest([price$, quantity$]).pipe(
  map(([price, quantity]) => price * quantity)
).subscribe(total => {
  this.total = total;
});

forkJoin — Wait for All to Complete

Emits once when all source Observables complete. Ideal for parallel HTTP requests.

import { forkJoin } from 'rxjs';

forkJoin({
  users: this.http.get<User[]>('/api/users'),
  posts: this.http.get<Post[]>('/api/posts'),
  comments: this.http.get<Comment[]>('/api/comments'),
}).subscribe(({ users, posts, comments }) => {
  this.users = users;
  this.posts = posts;
  this.comments = comments;
});

withLatestFrom — Secondary Values

Emits when the primary source emits, combining with the latest value from secondary sources.

import { withLatestFrom } from 'rxjs/operators';

this.clicks$.pipe(
  withLatestFrom(this.user$)
).subscribe(([clickEvent, user]) => {
  console.log('User', user.name, 'clicked at', clickEvent.clientX);
});

Joining Operator Comparison