Choosing the right RxJS Operator for Action Streams

If you use RxJS action streams (maybe for API calls?), you have probably paused over the same question I have:

Should this effect use mergeMap, switchMap, concatMap, or exhaustMap?

The honest answer is not “use this operator for GET and that operator for POST.” The right choice depends on what a repeated action means in your application.

I will use NgRx Effects in the first example because it is a familiar action-based API, but the ideas in this article are not tied to @ngrx/effects. They apply whenever an RxJS stream represents commands or events that start asynchronous work—for example, in an NgRx SignalStore method built with rxMethod, in a ComponentStore effect, or in your own action stream.

Let’s build a mental model that makes that choice much less mysterious.

The familiar effect

A typical effect listens for an action, calls a service, and maps the result to either a success or a failure action:

import { inject } from '@angular/core';
import { Actions, createEffect, ofType } from '@ngrx/effects';
import { catchError, map, of, switchMap } from 'rxjs';

export const loadTodos = createEffect(
  (
    actions$ = inject(Actions),
    todoService = inject(TodoService),
  ) =>
    actions$.pipe(
      ofType(TodosPageActions.load),
      switchMap(() =>
        todoService.getAll().pipe(
          map((todos) => TodosApiActions.loadSuccess({ todos })),
          catchError((error: unknown) =>
            of(TodosApiActions.loadFailure({ error })),
          ),
        ),
      ),
    ),
  { functional: true },
);

Modern NgRx supports functional effects, so the full example above uses one. From this point on, I will show only the relevant actions$.pipe(...) fragment. You can place the same pipeline inside a class-based effect, a functional effect, an rxMethod, or another RxJS-based action handler.

First: what are we flattening?

An action stream is an Observable. For every action, our service returns another Observable. We therefore have an Observable producing Observables—a higher-order Observable.

The four common flattening operators decide what happens when a new action arrives while the previous inner Observable is still running:

Operator	When another action arrives	Useful mental model
mergeMap	Subscribe to it too	Run in parallel
concatMap	Put it in a queue	One at a time, in order
switchMap	Unsubscribe from the previous inner Observable and switch to the new one	Latest subscription wins
exhaustMap	Ignore the new action while the current inner Observable is active	First one wins until completion

That last column is more useful than memorising a rule per HTTP verb.

mergeMap: run everything

mergeMap subscribes to every inner Observable as it arrives. Requests may run concurrently, and their responses may arrive in a different order.

This is a good fit when every operation matters and the operations are independent.

The danger is a race condition. If two responses update the same state, an older request can finish last and overwrite newer data.

concatMap: queue everything

concatMap waits for the current inner Observable to complete before subscribing to the next one.

Nothing is dropped, and the client starts operations in action order. The trade-off is waiting: one slow request blocks everything queued behind it.

This is often a good default for writes that must be applied in order.

switchMap: keep listening only to the latest

When a new action arrives, switchMap unsubscribes from the previous inner Observable and subscribes to the new one.

With Angular HttpClient, unsubscribing normally aborts the client-side request. But be careful with the wording: this does not prove that the server never received or processed it. A write may already have crossed that boundary.

This makes switchMap excellent for replaceable reads—search suggestions are the classic example—but risky as a general policy for writes.

exhaustMap: ignore repeated triggers

exhaustMap accepts the first action, then ignores new actions until the active inner Observable completes.

It is useful when repeated triggers are accidental or meaningless: double-clicking a login button, submitting the same form twice, or requesting the same deletion while it is already in progress.

Remember that ignored actions are not queued. They are gone.

Do not choose by HTTP verb alone

“Use switchMap for GET” sounds convenient, but it is too broad.

Ask these questions instead:

1. Does every action represent work that must happen?
2. Can two operations safely run at the same time?
3. Does order matter?
4. Is a newer action a replacement for an older one?
5. Should repeated actions be ignored while one is running?
6. Are those answers global, or only true for actions concerning the same entity?

The last question is where things get interesting.

Loading one replaceable collection

Suppose the store contains one visible list of todos. Every load action means “replace that list with the newest response.”

Here, switchMap is a natural fit:

actions$.pipe(
  ofType(TodosPageActions.load),
  switchMap(() =>
    todoService.getAll().pipe(
      map((todos) => TodosApiActions.loadSuccess({ todos })),
      catchError((error: unknown) =>
        of(TodosApiActions.loadFailure({ error })),
      ),
    ),
  ),
);

If the user refreshes twice, we usually care about the latest refresh, not both responses.

exhaustMap could also be correct—but it expresses a different product decision. It says that refresh clicks should be ignored until the current request completes. NgRx’s own effects guide uses this pattern in a collection-loading example. Neither operator is universally right; the UX decides.

Loading a collection with query parameters

Now imagine:

TodosPageActions.load({ completed: true });
TodosPageActions.load({ completed: false });

Should the first request be cancelled?

It depends on the state model.

If the store holds only the currently selected list, the second action replaces the first query. switchMap is still appropriate:

switchMap(({ completed }) =>
  todoService.getAll({ completed }).pipe(
    map((todos) =>
      TodosApiActions.loadSuccess({ completed, todos }),
    ),
    catchError((error: unknown) =>
      of(TodosApiActions.loadFailure({ completed, error })),
    ),
  ),
)

If the store caches a separate result for each query, both requests matter. mergeMap may be the better choice:

mergeMap(({ completed }) =>
  todoService.getAll({ completed }).pipe(
    map((todos) =>
      TodosApiActions.loadSuccess({ completed, todos }),
    ),
    catchError((error: unknown) =>
      of(TodosApiActions.loadFailure({ completed, error })),
    ),
  ),
)

Notice that the success action includes the query. Without it, the reducer may not know which cache entry to update.

The parameter alone did not determine the operator. The way we store the result did.

Loading individual entities

Consider these actions:

TodosPageActions.loadOne({ id: 1 });
TodosPageActions.loadOne({ id: 2 });

The requests concern different entities, so cancelling one because the other started would be surprising. mergeMap allows both to run:

mergeMap(({ id }) =>
  todoService.get(id).pipe(
    map((todo) => TodosApiActions.loadOneSuccess({ todo })),
    catchError((error: unknown) =>
      of(TodosApiActions.loadOneFailure({ id, error })),
    ),
  ),
)

But what about this?

TodosPageActions.loadOne({ id: 1 });
TodosPageActions.loadOne({ id: 1 });

Now the requests target the same entity. If they run in parallel, an older response could arrive last.

What we really want is:

• different IDs: run in parallel;
• the same ID: keep only the latest subscription.

That is keyed concurrency.

Keyed concurrency with groupBy

RxJS’s groupBy can split the action stream into one stream per ID. We can use switchMap inside each group and mergeMap across groups:

actions$.pipe(
  ofType(TodosPageActions.loadOne),
  groupBy(({ id }) => id),
  mergeMap((actionsForId$) =>
    actionsForId$.pipe(
      switchMap(({ id }) =>
        todoService.get(id).pipe(
          map((todo) =>
            TodosApiActions.loadOneSuccess({ todo }),
          ),
          catchError((error: unknown) =>
            of(TodosApiActions.loadOneFailure({ id, error })),
          ),
        ),
      ),
    ),
  ),
);

This gives us both behaviours:

• requests for different entities can run concurrently;
• a newer request for the same entity replaces the previous subscription.

If the discriminator is a query object, create a stable key from the meaningful fields. Do not casually use JSON.stringify unless key order and normalisation are under your control.

A groupBy lifetime warning

Application-level action streams are usually long-lived. Without a duration, every distinct group created by groupBy also remains available for the life of the subscription. That is harmless for a small, bounded set of keys, but an unbounded stream of unique IDs can retain an unbounded number of groups.

RxJS lets you provide a duration that closes an inactive group:

groupBy(
  ({ id }) => id,
  {
    duration: (group$) => group$.pipe(debounceTime(30_000)),
  },
)

This deserves more caution than it usually gets. If a group closes while one of its requests is still active, a later action with the same key creates a new group. The old request and the new group can then overlap, weakening the per-key guarantee.

Use a duration only when its timing is valid for your domain, or when that occasional overlap is acceptable. For strict, reusable keyed concurrency, prefer a thoroughly tested custom operator or a request-coordination layer with explicit lifecycle tracking. A five-second timeout copied from a blog post is not a correctness proof—I say that with affection as someone who once wanted it to be one.

Creating entities

For independent creates, every action normally matters. mergeMap lets them run concurrently:

actions$.pipe(
  ofType(TodosPageActions.add),
  mergeMap(({ draft }) =>
    todoService.add(draft).pipe(
      map((todo) => TodosApiActions.addSuccess({ todo })),
      catchError((error: unknown) =>
        of(TodosApiActions.addFailure({ draft, error })),
      ),
    ),
  ),
);

Use concatMap instead if creation order matters or the server requires serial execution.

Use exhaustMap if the UI represents a single submission and double-submission must be ignored. Disabling the submit button while the request is pending is still a good UX improvement, but the effect can provide a second line of defence.

switchMap is rarely the safe default for a create. Unsubscribing does not undo a record the server has already created.

Updating entities

This is where I would change a common recommendation: for writes to the same entity, “latest wins” on the client is not enough.

Imagine two edits:

TodosPageActions.update({ id: 1, changes: { title: 'First' } });
TodosPageActions.update({ id: 1, changes: { title: 'Second' } });

With switchMap, the first response may be ignored, but the server might still apply the first update after the second. The client then looks correct until the next reload reveals stale data.

A safer client-side policy is:

• different IDs: update in parallel;
• the same ID: queue updates in order.

actions$.pipe(
  ofType(TodosPageActions.update),
  groupBy(({ id }) => id),
  mergeMap((actionsForId$) =>
    actionsForId$.pipe(
      concatMap(({ id, changes }) =>
        todoService.update(id, changes).pipe(
          map((todo) =>
            TodosApiActions.updateSuccess({ todo }),
          ),
          catchError((error: unknown) =>
            of(TodosApiActions.updateFailure({ id, error })),
          ),
        ),
      ),
    ),
  ),
);

This preserves the order in which the client sends updates for each entity. It does not solve every distributed-systems problem: other clients can still write concurrently. If that matters, use server-side concurrency control such as version numbers, ETags, or conflict detection.

For high-frequency UI changes, another valid design is to debounce before sending writes, or to dispatch a “commit” action separately from local edits. That is a product and data-consistency decision, not merely an operator choice.

Deleting entities

For deletion, duplicate requests for the same ID are usually pointless, while deletions of different IDs can run in parallel.

That suggests exhaustMap per key:

actions$.pipe(
  ofType(TodosPageActions.delete),
  groupBy(({ id }) => id),
  mergeMap((actionsForId$) =>
    actionsForId$.pipe(
      exhaustMap(({ id }) =>
        todoService.delete(id).pipe(
          map(() =>
            TodosApiActions.deleteSuccess({ id }),
          ),
          catchError((error: unknown) =>
            of(TodosApiActions.deleteFailure({ id, error })),
          ),
        ),
      ),
    ),
  ),
);

A global exhaustMap would also suppress deletion of ID 2 while ID 1 is being deleted. That may be right for a single confirmation dialog, but it is too restrictive for many list UIs.

Also consider making deletion idempotent on the server. Client-side operator choices help the UX; server-side semantics protect the system.

A practical decision table

These are starting points, not laws:

Situation	Likely choice	Why
Latest search or filter replaces the previous one	switchMap	Old results are no longer useful
Refresh clicks should be ignored while loading	exhaustMap	Prevent duplicate in-flight loads
Independent reads or creates	mergeMap	All operations matter and may run together
Ordered writes	concatMap	Preserve client-side execution order
Latest read per entity	groupBy + switchMap	Replace only requests with the same key
Ordered writes per entity	groupBy + concatMap	Serialize the same entity, parallelise different ones
Ignore duplicate work per entity	groupBy + exhaustMap	Suppress repeats only for the same key

When uncertain, describe the desired behaviour in plain English first:

Run all of them, queue them, replace the previous one, or ignore the new one?

Then decide whether that rule applies to the whole action stream or separately per entity.

Should we hide this in a custom operator?

A custom operator can remove repetition, but it also hides the most important behaviour in the effect: concurrency.

Names such as PARALLEL, QUEUE, LATEST, and IGNORE_WHILE_BUSY are readable, but a general helper must also get error handling, typing, key lifetimes, cancellation, and per-key cleanup right. That is quite a contract.

My current preference is:

• keep the standard RxJS operator visible for simple effects;
• extract a keyed operator only after the pattern repeats;
• give it a behavioural name;
• test it with marble tests, including same-key and different-key actions;
• document exactly what happens when groups expire and requests outlive them.

Abstraction should make the concurrency policy clearer, not merely make the pipe shorter.

Final thought

The best flattening operator is not determined by GET, POST, PATCH, or DELETE, nor by whether you happen to be using NgRx Effects, SignalStore, or a home-grown action stream. It is determined by which actions matter, which may overlap, and what your state considers “the same piece of work.”

Once you ask those questions, the four operators stop feeling like RxJS trivia and start looking like four very precise product decisions. And that is much easier to reason about.