`race`

먼저 내보내는 후보가 이깁니다: 그 스트림 전체가 비춰지고, 다른 모든 후보는 그 자리에서 취소됩니다.

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

강의

캐시와 네트워크에 동시에 물어보고, 먼저 답하는 쪽을 취합니다. 다운로드 미러 셋을 시도하고, 가장 빠른 것을 남깁니다. FxEvents.race(candidates)는 모든 후보를 한꺼번에 구독하고, 어디서든 나온 첫 이벤트가 승부를 결정합니다: 그 이벤트가 전달되고, 진 후보들은 모두 즉시 취소됩니다 — 음소거가 아니라 취소입니다. 그들의 소켓은 닫히고, 작업은 멈추고, 이벤트는 일어나지 않습니다.

무엇이 이기는지 정확히 해 둡시다: 어디서든 나온 첫 이벤트가 승자를 고르고, 그때부터 race는 이긴 스트림을 온전히 비춥니다 — 이후 그 스트림이 만들어 내는 모든 이벤트가 흘러오고, 승자가 닫히면 race도 닫힙니다. 에러도 이길 수 있습니다: 빠르게 고장 난 엔드포인트가 느리지만 건강한 것을 이깁니다. 이것이 정확히 정직한 동작입니다 — 누가 먼저 답하는지 물었고, "실패했다"도 하나의 답이니까요. 느리고 불안정한 구간은 경주에 내보내기 전에 각 후보에 timeout/retry로 안전장치를 거세요.

한 번도 내보내지 않고 닫히는 후보는 조용히 탈락합니다. 모두가 그렇거나 — 후보 목록이 비어 있으면 — race는 빈 채로 닫힙니다. 이웃과의 취소 방향 차이에 주목하세요: switchMap은 연속된 스트림 중 더 오래된 쪽을 취소하고, race는 동시의 스트림 중 더 느린 쪽을 취소합니다. fxdart 이벤트 계층이며, Rx의 race/amb를 따랐습니다.

데모 1 · 캐시 대 네트워크, 진 쪽은 취소

데모 2 · 에러도 이길 수 있고, 빈 후보는 탈락한다

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

직접 해 보기

연습: 세 미러 중 가장 빠른 것.

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

관련 항목: switchMap — 더 느린 쪽 대신 더 오래된 쪽을 취소 · timeout — 경주 전에 각 후보에 한도 걸기 · fxEvents — 승자가 아니라 모두의 이벤트를 원할 때는 FxEvents.merge