Skip to content

mergeMap#

mergeMap() is a higher-order mapping operator used to handle scenarios where each value emitted by a source (outer) Observable triggers an asynchronous operation that returns another Observable (an inner Observable).

Here's how it works:

  1. It takes a value emitted by the source (outer) Observable.
  2. It uses that value and a function you provide to create a new inner Observable.
  3. It subscribes to this new inner Observable.
  4. Crucially (and unlike switchMap): If the source Observable emits a new value, mergeMap does not cancel or unsubscribe from any previous inner Observables that might still be running.
  5. It subscribes to the new inner Observable generated by the new source value and runs it concurrently with any other active inner Observables.
  6. It then merges the values emitted by all the active inner Observables into a single output stream. The order of the output values depends on when the inner Observables emit, not necessarily the order of the outer source emissions.

Think of it as spawning multiple asynchronous tasks based on incoming triggers and collecting all their results together as they complete, without cancelling anything.

At a glance

  • Signature: mergeMap(project, concurrent?) where concurrent caps how many inner Observables run at once
  • Use when: independent async operations should run in parallel and every result matters
  • Avoid when: order matters (use concatMap) or only the latest result matters (use switchMap)
  • Top gotcha: concurrency is unbounded by default; a burst of source values means a burst of parallel requests

Key Characteristics#

  • Higher-Order Mapping: Maps values from an outer Observable to inner Observables.
  • Concurrent Execution: Subscribes to and runs multiple inner Observables in parallel.
  • Merging Output: Combines emissions from all active inner Observables into a single output stream.
  • No Cancellation: Does not cancel previous inner operations when new outer values arrive.
  • Use Cases: Ideal when you need to perform multiple asynchronous actions concurrently based on source emissions and want the results from all of them. Useful when the order of completion isn't strictly important, and parallel processing is beneficial.

Minimal Example#

import { interval, map, mergeMap, take } from "rxjs";

// source emits 0, 1, one value per second
interval(1000)
  .pipe(
    take(2),
    mergeMap((n) =>
      interval(700).pipe(
        take(2),
        map((i) => `outer ${n} / inner ${i}`),
      ),
    ),
  )
  .subscribe(console.log);

// Output: every inner stream runs to completion, nothing is cancelled
// outer 0 / inner 0
// outer 0 / inner 1
// outer 1 / inner 0
// outer 1 / inner 1

Real-World Example Scenario#

Imagine you're working on a feature in an Angular application where a user can modify several pieces of data (e.g., multiple settings, or various documents in a list) and then click a single "Save All" button.

  • The click event triggers an action.
  • You get a list of items that need saving (e.g., ['settingA', 'settingB', 'documentX']).
  • For each item in the list, you need to make a separate API call (e.g., http.put('/api/settings/settingA', ...), http.put('/api/settings/settingB', ...), etc.).
  • You want these save operations to happen in parallel to make it faster. You don't want to wait for 'settingA' to finish saving before starting the save for 'settingB'.
  • You want to get feedback (like a success/error message) for each individual save operation as it completes.

mergeMap is perfect for this because it will take each item ID, trigger its corresponding API call (inner Observable), run all these API calls concurrently, and merge their results (e.g., success/error responses) into the output stream as they arrive.

Angular Example#

import { Component, inject, signal } from "@angular/core";
import { HttpClient } from "@angular/common/http";
import { catchError, from, map, mergeMap, of } from "rxjs";

interface SaveResult {
  id: string;
  success: boolean;
}

@Component({
  selector: "app-save-all",
  template: `
    <button (click)="saveAll()" [disabled]="saving()">
      {{ saving() ? "Saving..." : "Save All" }}
    </button>
    <ul>
      @for (result of results(); track result.id) {
        <li>{{ result.id }}: {{ result.success ? "saved" : "failed" }}</li>
      }
    </ul>
  `,
})
export class SaveAllComponent {
  private readonly http = inject(HttpClient);

  protected readonly saving = signal(false);
  protected readonly results = signal<SaveResult[]>([]);

  saveAll(): void {
    const ids = ["doc1", "settingA", "userPrefX"];
    this.saving.set(true);
    this.results.set([]);

    from(ids)
      .pipe(
        // all PUTs run concurrently; results arrive in completion order
        mergeMap((id) =>
          this.http.put(`/api/items/${id}`, { id }).pipe(
            map(() => ({ id, success: true })),
            // catch per item so one failure does not stop the others
            catchError(() => of({ id, success: false })),
          ),
        ),
      )
      .subscribe({
        next: (result) => this.results.update((r) => [...r, result]),
        complete: () => this.saving.set(false), // fires when ALL saves finish
      });
  }
}

How it works:

  1. from(ids) emits each item id one by one.
  2. mergeMap starts an HTTP PUT for each id immediately, without waiting for or cancelling the others. With three quick emissions there are three requests in flight at once.
  3. The inner map/catchError turn each response into a SaveResult, so one failed save cannot error the whole stream.
  4. next receives results in completion order, not input order. complete fires only after the source is done and every inner request has finished.
  5. To limit parallelism, pass a concurrency cap: mergeMap(project, 2) keeps at most two requests in flight and queues the rest.

Common Mistakes#

Using mergeMap for type-ahead search. Responses can arrive out of order, so results for an old term can overwrite results for the latest term. Use switchMap, which cancels stale requests.

No concurrency cap on large batches. from(thousandIds).pipe(mergeMap(fetch)) fires a thousand parallel requests. Pass the second argument, mergeMap(fetch, 4), to keep a bounded pool and queue the rest.

Expecting output order to match input order. Results are merged in completion order. If order matters, use concatMap.

Interview Q&A#

What does the second argument of mergeMap do?

It caps how many inner Observables are subscribed at once. Extra source values wait in a queue until a slot frees up. mergeMap(project, 1) behaves exactly like concatMap, which is a useful fact for follow-up questions.

Why can mergeMap cause race conditions in UIs?

Because nothing is cancelled and nothing is ordered: two in-flight requests can resolve in any order, and the UI shows whichever wrote last. For latest-wins UI state, switchMap is the fix; for strict ordering, concatMap.

When does the merged stream complete?

Only when the source has completed and every active inner Observable has completed. If any inner stream never completes (an interval, a long-lived Subject), the output never completes either, which is a common source of leaks.

Summary#

Use mergeMap() when you need to trigger multiple asynchronous operations based on incoming events/data and want them to run concurrently, collecting all their results as they finish. It's ideal for parallelism where cancellation of previous operations is not needed or desired.