`mergeMap`, `concatMap` & `exhaustMap`

Three more answers to "an event arrived while the last one is still running": run them all, run them in order, or ignore the new one.

FxEvents<R> FxEvents<T>.mergeMap<R>(Stream<R> Function(T) f, {int? concurrent}) FxEvents<R> FxEvents<T>.concatMap<R>(Stream<R> Function(T) f) FxEvents<R> FxEvents<T>.exhaustMap<R>(Stream<R> Function(T) f)

Lecture

Mapping an event to an inner stream — a request, an upload, a query — raises one question that a pull pipeline never has to answer: what happens when the next event arrives before the last inner stream has finished? There are exactly four sensible policies, and picking the wrong one is where most reactive bugs live. switchMap is the last-wins answer; these three are the other three.

mergeMap(f) runs every inner stream at once and interleaves their output in arrival order. Use it when every result matters and none supersedes another — uploading three files, fanning out to three services. With concurrent: n at most n run at a time and the rest wait in a queue, which is how you keep a fan-out from opening two hundred sockets.

concatMap(f) runs them strictly in order, each to completion before the next begins. Nothing overlaps and nothing is dropped, so a slow inner stream backs the whole chain up — that is the point when order is the correctness condition, as in "apply these edits in sequence".

exhaustMap(f) keeps the first and ignores the rest: while an inner stream is running, incoming events are dropped outright — not queued, not cancelled. This is the double-submit guard. A second tap on a button whose request is still in flight does nothing at all, which is exactly what you want when the request is POST /orders.

fxdart events layer, after Rx's flatMap, flatMap(maxConcurrent: 1) and exhaustMap. The first is called mergeMap here because flatMap already means iterable-flattening on the pull side.

Demo 1 · mergeMap — everything at once

Demo 2 · exhaustMap — the double-submit guard

Try it yourself

Exercise: concatMap's ordering, and a bounded fan-out.

import 'dart:async';

import 'package:fxdart/fxdart.dart';

/// Emits each (offsetMs, value) pair at its offset, closing at [closeMs].
Stream<T> timed<T>(List<(int, T)> events, int closeMs) {
  final c = StreamController<T>();
  for (final (ms, v) in events) {
    Timer(Duration(milliseconds: ms), () => c.add(v));
  }
  Timer(Duration(milliseconds: closeMs), c.close);
  return c.stream;
}

Future<void> main() async {
  // Exercise 1: concatMap keeps inner streams strictly in order.
  final pages = await fxEvents(Stream.fromIterable([1, 2, 3]))
      .concatMap((n) => timed([(120, 'page $n')], 150))
      .toList();
  print(pages); // [page 1, page 2, page 3] — one at a time, ~450ms

// Exercise 2: mergeMap with a concurrency cap — four jobs, two at a
  // time. Source order decides who STARTS; completion decides who is
  // emitted first.
  final jobs = await fxEvents(Stream.fromIterable([1, 2, 3, 4]))
      .mergeMap((n) => timed([(120, 'job $n')], 150), concurrent: 2)
      .toList();
  print(jobs); // [job 1, job 2, job 3, job 4] — in two waves

// The four policies, in one line each:
  //   mergeMap   — run them all, interleaved
  //   concatMap  — run them one after another, in order
  //   switchMap  — keep only the newest, cancel the rest
  //   exhaustMap — keep only the first, ignore the rest
}

Related: switchMap — the fourth policy: newest wins, the rest are cancelled · mapConcurrent — pull-side bounded fan-out, where results stay in order · debounce — often the better fix: stop the extra events before they become inner streams

mergeMap, concatMap & exhaustMap

Lecture

Demo 1 · mergeMap — everything at once

Demo 2 · exhaustMap — the double-submit guard

Try it yourself

`mergeMap`, `concatMap` & `exhaustMap`