Skip to content

concatMap#

concatMap() is a higher-order mapping operator that maps each value from a source (outer) Observable to an inner Observable, subscribes to it, but waits for that inner Observable to complete before moving on to map and subscribe to the inner Observable generated by the next value from the source.

Here's the flow:

  1. It takes the first value emitted by the source (outer) Observable.
  2. It uses that value and your project function to create the first inner Observable.
  3. It subscribes to this first inner Observable and emits its values.
  4. If the source Observable emits a second value while the first inner Observable is still running, concatMap holds onto that second value.
  5. Only when the first inner Observable completes does concatMap use the held second value to create and subscribe to the second inner Observable.
  6. This process continues, effectively creating a queue where inner Observables are executed one after another, strictly in the order dictated by the source Observable.

Think of it as processing tasks in a single-file line: the next task only starts once the current one is completely finished.

At a glance

  • Signature: concatMap(project), equivalent to mergeMap(project, 1)
  • Use when: operations must run one at a time, in source order: ordered writes, task queues
  • Avoid when: operations are independent and could run in parallel, or the source emits faster than inner streams complete
  • Top gotcha: an inner Observable that never completes blocks the queue forever

Key Characteristics#

  • Higher-Order Mapping: Maps values from an outer Observable to inner Observables.
  • Sequential Execution: Runs inner Observables one at a time, in order.
  • Waits for Completion: Does not subscribe to the next inner Observable until the previous one completes.
  • Preserves Order: Guarantees that the output values maintain the order corresponding to the source emissions.
  • Use Cases: Ideal when the order of operations is crucial, or when you need to ensure one asynchronous task finishes before the next begins (e.g., to avoid race conditions, maintain data integrity, or process items sequentially). Also useful for implicit rate-limiting when you don't want concurrent requests.

Minimal Example#

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

interval(300)
  .pipe(
    take(3), // source emits 0, 1, 2 quickly
    concatMap((n) =>
      interval(1000).pipe(
        take(2),
        map((i) => `outer ${n} / inner ${i}`),
      ),
    ),
  )
  .subscribe(console.log);

// Inner streams never overlap; order is preserved:
// outer 0 / inner 0
// outer 0 / inner 1
// outer 1 / inner 0
// outer 1 / inner 1
// outer 2 / inner 0
// outer 2 / inner 1

Real-World Example Scenario#

Imagine you're building a feature where a user can trigger several updates that need to be applied to a database or configuration file in a specific order to maintain consistency.

Scenario: A user is rapidly clicking buttons to add different items to a configuration profile ("Add Feature A", "Enable Setting B", "Add User C"). Each click triggers an API call to update the profile. If these updates happened concurrently (mergeMap), they might interfere with each other or lead to an inconsistent final state depending on server response times. If you used switchMap, clicking "Enable Setting B" might cancel the "Add Feature A" request if it was still pending.

You want to ensure the updates are applied strictly in the order the user clicked:

  1. Wait for "Add Feature A" API call to complete successfully.
  2. Then execute the "Enable Setting B" API call and wait for it to complete.
  3. Then execute the "Add User C" API call.

concatMap enforces this sequential processing.

Angular Example#

import { Component, inject, signal } from "@angular/core";
import { HttpClient } from "@angular/common/http";
import { takeUntilDestroyed } from "@angular/core/rxjs-interop";
import { Subject, catchError, concatMap, map, of } from "rxjs";

@Component({
  selector: "app-sequential-updates",
  template: `
    <p>Click rapidly: updates still run strictly one at a time, in order.</p>
    <button (click)="queue('add-feature')">Add Feature</button>
    <button (click)="queue('enable-setting')">Enable Setting</button>
    <button (click)="queue('add-user')">Add User</button>

    <ul>
      @for (entry of log(); track $index) {
        <li>{{ entry }}</li>
      }
    </ul>
  `,
})
export class SequentialUpdatesComponent {
  private readonly http = inject(HttpClient);
  private readonly actions$ = new Subject<string>();

  protected readonly log = signal<string[]>([]);

  constructor() {
    this.actions$
      .pipe(
        // one update at a time, in click order; later clicks wait in a queue
        concatMap((action) =>
          this.http.post(`/api/profile/${action}`, {}).pipe(
            map(() => `${action}: done`),
            // catch per action so one failure does not kill the queue
            catchError(() => of(`${action}: failed`)),
          ),
        ),
        takeUntilDestroyed(),
      )
      .subscribe((entry) => this.log.update((l) => [...l, entry]));
  }

  queue(action: string): void {
    this.actions$.next(action);
  }
}

How it works:

  1. Button clicks push action names into a Subject, which is the source stream.
  2. concatMap maps each action to an HTTP POST, but subscribes to the next one only after the current one completes. Rapid clicks queue up instead of firing in parallel.
  3. The per-action catchError converts a failure into a value, so the queue keeps processing later actions.
  4. takeUntilDestroyed() ends the subscription with the component, so no leak survives navigation.

Common Mistakes#

An inner Observable that never completes. concatMap waits for completion before starting the next item. Map to something that completes (an HTTP call, of(...), a stream with take/first), or the queue stalls forever.

Using it on high-frequency streams. If the source emits faster than inner streams finish, the queue grows without bound. Consider switchMap (drop stale), exhaustMap (ignore while busy), or a bounded mergeMap.

Skipping per-item error handling. An uncaught inner error tears down the entire chain, losing every queued item. Put catchError on the inner Observable, as in the example.

Interview Q&A#

How does concatMap relate to mergeMap?

concatMap(project) is exactly mergeMap(project, 1): a merge with concurrency capped at one. That framing usually impresses interviewers because it shows you understand the concurrency model rather than memorizing operator names.

What happens if the source emits while an inner Observable is still running?

The value is buffered. concatMap subscribes to its inner Observable only after the current one completes, so order is preserved at the cost of latency and a growing queue under load.

When would you pick exhaustMap over concatMap for button clicks?

concatMap queues every click, so five impatient clicks cause five sequential saves. exhaustMap ignores clicks while a save is in flight, which is usually what a submit button wants.

Summary#

Use concatMap when the order of execution matters and you need to ensure that asynchronous operations triggered by a stream of events happen sequentially, one completing before the next one begins. It's your tool for enforcing order in asynchronous workflows.