`race`

El primer candidato en emitir gana: su stream entero se refleja, y todos los demás candidatos se cancelan en el acto.

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

Lección

Pregunta a la caché y a la red a la vez; quédate con quien responda primero. Prueba tres mirrors de descarga; conserva el más rápido. FxEvents.race(candidates) se suscribe a todos los candidatos a la vez, y el primer evento en cualquier parte lo decide: ese evento se entrega, y todo candidato perdedor se cancela de inmediato — no se silencia, se cancela. Sus sockets se cierran, su trabajo se detiene, sus eventos nunca ocurren.

Sé preciso sobre qué gana: el primer evento en cualquier parte elige al ganador, y a partir de ahí la carrera refleja el stream ganador al completo — cada evento posterior que produzca fluye a través, y la carrera se cierra cuando el ganador se cierra. Un error también puede ganar: un endpoint roto pero rápido vence a uno sano pero lento, que es exactamente el comportamiento honesto — preguntaste quién responde primero, y «falló» es una respuesta. Protege los campos lentos y poco fiables con timeout/retry en cada candidato antes de ponerlos a correr.

Los candidatos que se cierran sin emitir nunca se retiran en silencio; si todos lo hacen — o la lista de candidatos está vacía — la carrera se cierra vacía. Fíjate en la dirección de la cancelación frente a su vecino: switchMap cancela el más viejo de streams sucesivos, race cancela el más lento de streams simultáneos. Capa de eventos de fxdart, según el race/amb de Rx.

Demo 1 · Caché contra red, perdedores cancelados

Demo 2 · Los errores pueden ganar, los candidatos vacíos se retiran

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());
  // []
}

Pruébalo tú

Ejercicio: el más rápido de tres 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.
}

Relacionado: switchMap — cancelar el más viejo en lugar del más lento · timeout — acota cada candidato antes de la carrera · fxEvents — FxEvents.merge, cuando quieres los eventos de todos, no un ganador