`waitAll`, `zip` 및 관련 함수

여러 스트림을 하나로 합치기: 전부 기다리거나, 인덱스로 짝짓거나, 차례로 재생하거나, 경주시키거나.

static FxEvents<List<T>> FxEvents.waitAll<T>(Iterable<Stream<T>> sources) static FxEvents<R> FxEvents.zip<T, R>(Iterable<Stream<T>> sources, R Function(List<T>) combine) static FxEvents<List<T>> FxEvents.combineLatestAll<T>(Iterable<Stream<T>> sources) static FxEvents<T> FxEvents.concat<T>(Iterable<Stream<T>> sources) FxEvents<R> FxEvents<T>.zipWith<U, R>(Stream<U> other, R Function(T, U) combine) FxEvents<T> FxEvents<T>.followedBy(Stream<T> next) // chain (events) FxEvents<T> FxEvents<T>.mergeWith(Stream<T> other) FxEvents<T> FxEvents<T>.raceWith(Stream<T> other)

강의

FxEvents.waitAll(sources)는 스트림을 위한 Future.wait입니다. 모든 소스가 닫히고 나면 정확히 하나의 이벤트 — 각 소스의 마지막 값을 소스 순서대로 담은 리스트 — 를 내보내고, 자신도 닫습니다. 대시보드 사례가 그것입니다: 패널 셋이 각자 로드되고, 가장 느린 하나가 들어오면 화면이 그려집니다. 한 번도 내보내지 않고 닫힌 소스가 있으면 보고할 완전한 결과가 없으므로, 아무것도 내보내지 않습니다.

FxEvents.zip은 소스를 인덱스로 짝짓습니다: 모든 소스의 1번째끼리, 그다음 모든 소스의 2번째끼리, 이렇게요. 앞서가는 소스는 가장 느린 소스가 따라올 때까지 버퍼에 담기고, 닫힌 소스의 버퍼가 비는 순간 결과도 닫힙니다 — 더 이상 짝을 만들 수 없으니까요. zipWith는 두 소스 형태이고, 리스트 기반 정적 메서드와 달리 서로 다른 타입을 짝지을 수 있습니다.

zip과 combineLatest는 나란히 두고 볼 가치가 있습니다. "두 스트림 합치기"의 두 절반인데 사람들이 끊임없이 잘못 고르거든요. zip은 위치로 짝짓습니다: A의 3번째는 아무리 오래 걸려도 언제나 B의 3번째를 만납니다. combineLatest는 시간으로 짝짓습니다: 어느 쪽 이벤트든 상대가 지금 쥐고 있는 값과 함께 다시 나가므로, 한 소스가 여러 출력에 등장하고 다른 소스는 하나에도 못 낄 수 있습니다. combineLatestAll이 그 N항 형태입니다.

나머지는 합치기가 아니라 차례 세우기입니다. FxEvents.concat은 각 소스를 완료까지 재생한 뒤 다음을 시작합니다 — followedBy가 그 두 소스 형태이고, Dart 자신의 Iterable.followedBy에서 이름을 땄습니다. mergeWith와 raceWith는 FxEvents.merge, FxEvents.race의 인스턴스 형태입니다. fxdart 이벤트 레이어, Rx의 forkJoin, zip, combineLatestList, concat을 따랐습니다.

데모 1 · 모든 패널 기다리기

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 {
  // A dashboard that cannot render until all three panels have loaded.
  final profile = timed([(80, 'profile')], 120);
  final orders = timed([(240, 'orders')], 280);
  final prefs = timed([(40, 'prefs')], 80);

final watch = Stopwatch()..start();
  final ready = await FxEvents.waitAll([profile, orders, prefs]).toList();
  watch.stop();

print(ready); // [[profile, orders, prefs]]
  print('all in after ~${(watch.elapsedMilliseconds / 100).round() * 100}ms');
  // ONE event, holding each source's LAST value in source order, once
  // every source has closed — the stream counterpart of Future.wait.
  // A source that closes without ever emitting means no result at all.
}

데모 2 · 위치로, 또는 시간으로

import 'dart:async';

import 'package:fxdart/fxdart.dart';

Future<void> main() async {
  // zip pairs sources up by INDEX: 1st with 1st, 2nd with 2nd.
  final paired = await FxEvents.zip<int, String>([
    Stream.fromIterable([1, 2, 3]),
    Stream.fromIterable([10, 20]),
  ], (values) => values.join('+'))
      .toList();
  print(paired); // [1+10, 2+20] — the unmatched 3 is never emitted

// zipWith is the two-source form, and it can cross types.
  final labelled = await fxEvents(Stream.fromIterable([1, 2]))
      .zipWith<String, String>(
          Stream.fromIterable(['a', 'b', 'c']), (n, s) => '$n$s')
      .toList();
  print(labelled); // [1a, 2b]

// combineLatestAll re-emits EVERY source's latest on every event.
  final a = timed([(0, 'a1'), (100, 'a2')], 300);
  final b = timed([(50, 'b1')], 300);
  print(await FxEvents.combineLatestAll([a, b]).toList());
  // [[a1, b1], [a2, b1]] — zip pairs by position, combineLatest by time.
}

직접 해 보기

연습: 차례 세우기 — concat, followedBy, raceWith, mergeWith.

import 'dart:async';

import 'package:fxdart/fxdart.dart';

Future<void> main() async {
  // Exercise 1: concat plays sources through in order, one at a time.
  print(await FxEvents.concat([
    Stream.fromIterable([1, 2]),
    Stream.fromIterable([3]),
  ]).toList()); // [1, 2, 3]

// Exercise 2: followedBy is the two-source form — cache first, then
  // the network.
  print(await fxEvents(Stream.fromIterable(['cache']))
      .followedBy(Stream.fromIterable(['network']))
      .toList()); // [cache, network]

// Exercise 3: raceWith keeps whichever source speaks first and
  // cancels the loser outright.
  print(await fxEvents(timed([(0, 'fast')], 60))
      .raceWith(timed([(300, 'slow')], 360))
      .toList()); // [fast]

// Exercise 4: mergeWith interleaves both in arrival order.
  print(await fxEvents(timed([(0, 'a'), (100, 'c')], 200))
      .mergeWith(timed([(50, 'b')], 200))
      .toList()); // [a, b, c]
}

관련 항목: combineLatest — 시간 기준 짝짓기, UI 상태에는 보통 이쪽 · race — 먼저 말하는 소스가 이기고 나머지는 취소 · zip — Iterable을 인덱스로 짝짓는 풀 레이어의 원본

waitAll, zip 및 관련 함수

강의

데모 1 · 모든 패널 기다리기

데모 2 · 위치로, 또는 시간으로

직접 해 보기

`waitAll`, `zip` 및 관련 함수