Debounced search

Time first, then latest-wins, then a typed parse. One chain. This is a tie with RxDart on purpose — the point is you do not need a second package.

fxEvents(keystrokes) .debounce(Duration) // wait for quiet .switchMap(search) // latest query wins .mapEither((r, hit) => parse) // failure as a Left

Lecture

A search box is not a list. Keystrokes arrive when they arrive, a burst should collapse to the last query, and a request that is still in flight when the next query lands should be cancelled — otherwise a slow "da" can overwrite a fast "dart". That is a push job: fxEvents + debounce + switchMap.

The RxDart comparison of the same job is a tie. rxdart's debounceTime is the same idea. FxDart's claim here is not speed — you cannot beat waiting — it is that the events layer lives in the same import as the pull pipelines and the typed-error system, so the next step (parse the hit, keep a Left instead of throwing) does not start a new library.

mapEither runs each delivered result in a raise scope. A bad payload becomes a Left; later queries still arrive. Put attempt on the source only when a throw should become a value — and if the chain also retries, attempt goes after retryOn, never before.

Demo 1 · wait for the typing to go quiet

The keystroke schedule is simulated: a burst, a pause, one more query. debounce(160ms) emits twice — fxd and fxdart — the same contract as the comparison page.

Demo 2 · latest query wins, then a typed parse

switchMap starts a search per query and cancels the previous inner stream. mapEither then names a missing hit as a Left instead of throwing. Both searches start; only the latest result is delivered.

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

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

final out = await fxEvents(queries)
      .switchMap(search)
      .mapEither<String, String>(
        (r, hit) => r.ensureNotNull(
          hit.startsWith('hits') ? hit : null,
          () => 'empty: $hit',
        ),
      )
      .toList();

print('started:   $started'); // [da, dart]
  print('delivered: $out'); // [Right(hits for "dart")]
}

Related: which surface — why this is push · fxEvents · debounce · switchMap · mapEither · bounded concurrent fetch — the I/O job · RxDart vs FxDart: debounce