`switchMap`

Mapea cada evento a un stream interno y refleja solo el más nuevo — un evento fresco cancela el stream interno anterior en pleno vuelo.

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

Lección

El bug de la caja de búsqueda que toda UI ha enviado al menos una vez: el usuario teclea da, luego dart; la petición de da es más lenta; sus resultados obsoletos llegan los últimos y pisan los buenos. La solución no es ignorar la respuesta vieja sino hacerla imposible: switchMap(f) mapea cada evento a un stream interno y, en el momento en que llega un evento más nuevo, cancela de raíz el stream interno anterior. Solo el stream interno más nuevo se refleja aguas abajo.

La cancelación por novedad es una política, y es la correcta exactamente cuando el trabajo viejo se vuelve inútil en cuanto existe entrada más nueva — búsqueda, autocompletado, navegación, «cargar los detalles de la fila seleccionada». Es la equivocada cuando la salida de cada stream interno importa (una subida por archivo, digamos) — ese es un trabajo de fan-out para el mapConcurrent del lado pull, donde nada se cancela.

Semántica en los bordes: la cadena se cierra cuando la fuente se ha cerrado y el último stream interno completa — que la fuente termine nunca corta el trabajo que ya está en pantalla. Un mapper que lanza se convierte en un evento de error y la fuente sigue adelante. Capa de eventos de fxdart, según el switchMap de Rx.

Demo 1 · La caja de búsqueda, arreglada

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 · El último stream interno termina de decir lo suyo

Pruébalo tú

Ejercicio: autocompletado que no puede mostrar sugerencias obsoletas.

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.
}

Relacionado: debounce — su pareja natural aguas arriba: menos consultas de entrada, cero resultados obsoletos de salida · race — cancelación entre streams hermanos en lugar de sucesivos · mapConcurrent — cuando cada resultado importa, fan-out del lado pull