`switchMap`

Maps each event to an inner stream and mirrors only the newest one — a fresh event cancels the previous inner stream mid-flight.

FxEvents<R> FxEvents<T>.switchMap<R>(Stream<R> Function(T a) f) // chain (events)

Lecture

The search-box bug every UI has shipped at least once: the user types da, then dart; the da request is slower; its stale results arrive last and overwrite the good ones. The fix is not to ignore the old response but to make it impossible: switchMap(f) maps each event to an inner stream and, the moment a newer event arrives, cancels the previous inner stream outright. Only the newest inner stream is ever mirrored downstream.

Cancellation-by-newer is a policy, and it is the right one exactly when older work becomes worthless once newer input exists — search, autocomplete, navigation, "load details for the selected row". It is the wrong one when every inner stream's output matters (an upload per file, say) — that is a fan-out job for the pull side's mapConcurrent, where nothing gets cancelled.

Semantics at the edges: the chain closes when the source has closed and the last inner stream completes — the source ending never cuts short work already on screen. A mapper that throws becomes an error event and the source keeps going. fxdart events layer, after Rx's switchMap.

Demo 1 · The search box, fixed

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 {
  // A search box: the user refines the query while the first search is
  // still in flight (each search takes ~150ms).
  final queries = timed([(0, 'da'), (60, 'dart')], 400);
  final started = <String>[];

Stream<String> search(String q) {
    started.add(q);
    return timed([(150, 'results for "$q"')], 200);
  }

final out = await fxEvents(queries).switchMap(search).toList();

print('started:   $started'); // [da, dart]
  print('delivered: $out'); // [results for "dart"]
  // Both searches started — but 'dart' arrived while 'da' was in flight,
  // so the older inner stream was CANCELLED mid-request. No stale results
  // can ever overwrite newer ones.
}

Demo 2 · The last inner stream finishes its say

Try it yourself

Exercise: autocomplete that cannot show stale suggestions.

import 'dart:async';

import 'package:fxdart/fxdart.dart';

Future<void> main() async {
  // Autocomplete: every keystroke kicks off a ~100ms suggestion fetch.
  final keys = timed([(0, 's'), (40, 'sw'), (80, 'swi')], 300);

Stream<String> suggest(String q) => timed([(100, '$q…suggestions')], 130);

// TODO: only the NEWEST query's suggestions should reach the screen —
  // replace asyncExpand (which dutifully shows every stale result) with
  //   fxEvents(keys).switchMap(suggest).toList()
  final out = await keys.asyncExpand(suggest).toList();

print(out);
  // currently every query's suggestions — want [swi…suggestions]:
  // 's' and 'sw' are superseded before their fetch comes back.
}

Related: debounce — its natural upstream partner: fewer queries in, no stale results out · race — cancellation between sibling streams instead of successive ones · mapConcurrent — when every result matters, pull-side fan-out