Race two mirrors

Toss-up async

Requirement

The same payload is available from two mirrors: the EU mirror answers in 60 ms, the US mirror in 180 ms. Fetch it as fast as possible and make sure the slow fetch does not run to completion — prove it by counting completed fetches well after the loser's deadline. The mirrors are simulated in the code as cancellable streams; both versions must print the lines shown under Expected output.

Expected output

winner: payload from eu-mirror
completed fetches: 1
loser fetch completed: false

Side by side

RxDart

import 'dart:async';

import 'package:rxdart/rxdart.dart';

final completed = <String>[];

/// A mirror fetch as a cancellable stream: the work is a timer that only
/// runs while someone is listening — cancelling really stops the fetch.
Stream<String> mirror(String name, int ms) {
  final c = StreamController<String>();
  Timer? t;
  c.onListen = () {
    t = Timer(Duration(milliseconds: ms), () {
      completed.add(name);
      c.add('payload from $name');
      c.close();
    });
  };
  c.onCancel = () => t?.cancel();
  return c.stream;
}

Future<void> main() async {
  // Race both mirrors: first event wins, the loser is CANCELLED mid-flight.
  final winner = await Rx.race(
      [mirror('eu-mirror', 60), mirror('us-mirror', 180)]).first;

// Wait long past the loser's 180 ms — its timer was cancelled at ~60 ms.
  await Future.delayed(const Duration(milliseconds: 540));

print('winner: $winner');
  print('completed fetches: ${completed.length}');
  print('loser fetch completed: ${completed.contains('us-mirror')}');
}

FxDart

import 'dart:async';

import 'package:fxdart/fxdart.dart';

final completed = <String>[];

Future<void> main() async {
  // Race both mirrors: first event wins, the loser is CANCELLED mid-flight.
  final winner = await FxEvents.race(
      [mirror('eu-mirror', 60), mirror('us-mirror', 180)]).head();

// Wait long past the loser's 180 ms — its timer was cancelled at ~60 ms.
  await Future.delayed(const Duration(milliseconds: 540));

print('winner: $winner');
  print('completed fetches: ${completed.length}');
  print('loser fetch completed: ${completed.contains('us-mirror')}');
}

Why they differ

They don't. Racing is a push idea — subscribe to everything, keep whoever speaks first, cancel the rest — and FxEvents.race is exactly that, matching Rx.race move for move: both mirrors are genuinely in flight, and the moment the EU mirror emits at 60 ms the US subscription is cancelled, its onCancel fires, and the pending timer dies. Both panels prove it the same way — the completed-fetch count is still 1 long after the loser's 180 ms deadline. The work was stopped, not just ignored, on both sides.

A pull chain can only decline to start the backup fetch, never cancel one in flight; the events layer absorbs the Rx approach instead: a thin wrapper chain over plain Streams that collides with nothing, rxdart included. RxDart's operator catalog remains far larger — fxdart keeps the events core small and hands the winner's per-value processing to the typed pull side via .pull(). For "first responder wins, losers are cancelled", the two are operator-for-operator equivalent: a tie.

Benchmark

No benchmark for this example. This operator is defined by elapsed time, not by work done: both sides wait out the same windows, so a bar would be measuring Dart's timers rather than either library. Shrinking the windows does not help either — how many values a burst produces would then depend on how fast the machine is, and the two sides could no longer be checked against an identical result. The examples built on timing operators are left unmeasured for that reason; the ones whose delays merely stand in for I/O are benchmarked with those delays set to zero.