`waitAll`, `zip` & friends

Combining many streams into one: wait for all of them, pair them up by index, play them in sequence, or let them race.

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)

Lecture

FxEvents.waitAll(sources) is Future.wait for streams. It emits exactly one event — a list holding each source's last value, in source order — once every source has closed, and then closes itself. That is the dashboard case: three panels load independently, and the screen renders when the slowest one is in. A source that closes without ever emitting means there is no complete result to report, so nothing is emitted at all.

FxEvents.zip pairs sources by index: every source's 1st event together, then every source's 2nd, and so on. Whichever source runs ahead is buffered until the slowest catches up, and the result closes as soon as a closed source runs out of buffer — no further pair can ever be formed. zipWith is the two-source form, and unlike the list-based static it can pair different types.

It is worth holding zip and combineLatest side by side, because they are the two halves of "combine two streams" and people reach for the wrong one constantly. zip pairs by position: the 3rd of A always meets the 3rd of B, however long that takes. combineLatest pairs by time: any event re-emits with whatever the other side happens to hold right now, so one source can appear in many outputs and another in none. combineLatestAll is its N-ary form.

The rest are sequencing rather than combining. FxEvents.concat plays each source through to completion before starting the next — followedBy is its two-source form, named after Dart's own Iterable.followedBy. mergeWith and raceWith are the instance forms of FxEvents.merge and FxEvents.race. fxdart events layer, after Rx's forkJoin, zip, combineLatestList and concat.

Demo 1 · Waiting for every panel

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

Demo 2 · By position, or by time

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

Try it yourself

Exercise: sequencing — 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]
}

Related: combineLatest — the by-time pairing, and the one you usually want for UI state · race — first source to speak wins, the others are cancelled · zip — the pull-layer original, pairing Iterables by index

waitAll, zip & friends

Lecture

Demo 1 · Waiting for every panel

Demo 2 · By position, or by time

Try it yourself

`waitAll`, `zip` & friends