두 미러 경주시키기

우열 없음 async

요구사항

같은 페이로드를 두 미러에서 받을 수 있습니다: EU 미러는 60 ms에, US 미러는 180 ms에 응답합니다. 최대한 빨리 가져오되, 느린 fetch가 끝까지 실행되지 않도록 하세요 — 패자의 마감 시간이 한참 지난 뒤 완료된 fetch 수를 세어 증명합니다. 미러들은 코드에 취소 가능한 스트림으로 시뮬레이션되어 있습니다; 두 버전 모두 예상 출력 아래에 표시된 줄들을 출력해야 합니다.

예상 출력

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

나란히 보기

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')}');
}

차이가 나는 이유

다르지 않습니다. 경주는 push의 아이디어입니다 — 전부 구독하고, 먼저 말하는 쪽을 지키고, 나머지는 취소합니다 — 그리고 FxEvents.race가 정확히 그것이며, Rx.race와 수 하나하나까지 맞아떨어집니다: 두 미러가 진짜로 동시에 진행되고, EU 미러가 60 ms에 값을 내보내는 순간 US 구독이 취소되어 onCancel이 발동하고, 대기 중이던 타이머가 죽습니다. 두 패널 모두 같은 방식으로 증명합니다 — 패자의 180 ms 마감이 한참 지난 뒤에도 완료된 fetch 수가 여전히 1입니다. 양쪽 모두에서 일은 그저 무시된 것이 아니라 중단되었습니다.

pull 체인은 백업 fetch의 시작을 거절할 수 있을 뿐, 진행 중인 것을 취소하지는 못합니다; 대신 이벤트 레이어가 Rx의 접근을 흡수합니다: 평범한 Stream 위의 얇은 래퍼 체인이라 rxdart를 포함해 어떤 것과도 충돌하지 않습니다. RxDart의 연산자 카탈로그는 여전히 훨씬 큽니다 — fxdart는 이벤트 코어를 작게 유지하고, 승자의 값별 처리는 .pull()을 통해 타입 있는 pull 쪽에 넘깁니다. "먼저 응답하는 쪽이 이기고, 패자는 취소된다"에서 둘은 연산자 대 연산자로 동등합니다: 무승부입니다.

벤치마크

이 예제에는 벤치마크가 없습니다. 이 연산자는 처리한 일의 양이 아니라 흘러간 시간으로 정의됩니다. 양쪽 모두 똑같은 시간 창을 기다리므로, 막대는 두 라이브러리가 아니라 Dart의 타이머를 재게 됩니다. 시간 창을 줄여도 해결되지 않습니다 — 그러면 한 번의 이벤트 폭주에서 몇 개의 값이 나오는지가 머신 속도에 좌우되고, 두 구현이 같은 결과를 냈는지 대조할 수 없게 되기 때문입니다. 타이밍 연산자로 만들어진 예제들을 측정하지 않고 두는 이유가 이것입니다. 지연이 그저 I/O를 흉내 낼 뿐인 예제들은 그 지연을 0으로 두고 벤치마크합니다.