`race`

The first candidate to emit wins: its whole stream is mirrored, and every other candidate is cancelled on the spot.

static FxEvents<T> FxEvents.race<T>(Iterable<Stream<T>> candidates)

Lecture

Ask the cache and the network at the same time; take whichever answers first. Try three download mirrors; keep the fastest. FxEvents.race(candidates) subscribes to every candidate at once, and the first event anywhere decides it: that event is delivered, and every losing candidate is cancelled immediately — not muted, cancelled. Their sockets close, their work stops, their events never happen.

Be precise about what wins: the first event anywhere picks the winner, and from then on the race mirrors the winning stream in full — every later event it produces flows through, and the race closes when the winner closes. An error can win too: a fast broken endpoint beats a slow healthy one, which is exactly the honest behavior — you asked who answers first, and "it failed" is an answer. Guard slow-and-flaky fields with timeout/retry on each candidate before racing them.

Candidates that close without ever emitting drop out silently; if all of them do — or the candidate list is empty — the race closes empty. Note the direction of cancellation versus its neighbor: switchMap cancels the older of successive streams, race cancels the slower of simultaneous ones. fxdart events layer, after Rx's race/amb.

Demo 1 · Cache vs network, losers cancelled

Demo 2 · Errors can win, empty candidates drop 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 {
  // An ERROR can win the race too: the first event decides, whether it
  // is a value or a failure. A fast broken endpoint beats a slow good one.
  final failing = timed([(30, 'x')], 60)
      .asyncMap((_) => Future<String>.error(StateError('mirror down')));
  final slow = timed([(200, 'slow but fine')], 240);
  try {
    await FxEvents.race([failing, slow]).toList();
  } on StateError catch (e) {
    print('race lost to an error: ${e.message}');
  }

// Candidates that close WITHOUT emitting simply drop out; when all of
  // them do (or the field is empty), the race closes empty.
  print(await FxEvents.race<int>([timed([], 30), timed([], 50)]).toList());
  // []
  print(await FxEvents.race<int>(const []).toList());
  // []
}

Try it yourself

Exercise: fastest of three mirrors.

import 'dart:async';

import 'package:fxdart/fxdart.dart';

/// A download mirror that answers after [ms] milliseconds.
Stream<String> mirror(String name, int ms) =>
    timed([(ms, 'downloaded from $name')], ms + 40);

Future<void> main() async {
  // TODO: race the three mirrors — the first byte wins and the other two
  // connections are cancelled immediately:
  //   FxEvents.race([mirror('eu', 120), mirror('us', 60), mirror('ap', 180)])
  //       .toList()
  final out = await mirror('us', 60).toList();

print(out);
  // want [downloaded from us] — and with race, the eu/ap sockets are
  // closed the moment us answers, instead of running to completion.
}

Related: switchMap — cancelling the older instead of the slower · timeout — bound each candidate before the race · fxEvents — FxEvents.merge, when you want everyone's events, not a winner