`waitAll`, `zip` & friends

Combinar muchos streams en uno: esperarlos a todos, emparejarlos por índice, reproducirlos en secuencia o dejar que compitan.

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)

Lección

FxEvents.waitAll(sources) es el Future.wait de los streams. Emite exactamente un evento — una lista con el último valor de cada fuente, en el orden de las fuentes — en cuanto todas se han cerrado, y luego se cierra él. Ese es el caso del panel de control: tres paneles cargan de forma independiente y la pantalla se dibuja cuando entra el más lento. Una fuente que se cierra sin haber emitido nada significa que no hay resultado completo que informar, así que no se emite nada en absoluto.

FxEvents.zip empareja las fuentes por índice: el 1.er evento de cada fuente junto, luego el 2.º de cada una, y así sucesivamente. La fuente que va por delante se almacena hasta que la más lenta la alcanza, y el resultado se cierra en cuanto una fuente cerrada se queda sin búfer: ya no puede formarse ningún par más. zipWith es la forma de dos fuentes y, a diferencia de la estática basada en listas, puede emparejar tipos distintos.

Merece la pena poner zip y combineLatest uno al lado del otro, porque son las dos mitades de «combinar dos streams» y la gente echa mano del equivocado constantemente. zip empareja por posición: el 3.º de A siempre se encuentra con el 3.º de B, tarde lo que tarde. combineLatest empareja por tiempo: cualquier evento vuelve a emitir con lo que el otro lado tenga en ese momento, así que una fuente puede aparecer en muchas salidas y otra en ninguna. combineLatestAll es su forma N-aria.

Los demás son secuenciación más que combinación. FxEvents.concat reproduce cada fuente hasta el final antes de empezar la siguiente; followedBy es su forma de dos fuentes, llamada así por el propio Iterable.followedBy de Dart. mergeWith y raceWith son las formas de instancia de FxEvents.merge y FxEvents.race. Capa de eventos de fxdart, siguiendo a forkJoin, zip, combineLatestList y concat de Rx.

Demo 1 · Esperar a cada 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 · Por posición o por tiempo

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

Pruébalo tú

Ejercicio: secuenciación — 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]
}

Relacionado: combineLatest — el emparejado por tiempo, y el que normalmente quieres para el estado de la UI · race — gana la primera fuente que habla, las demás se cancelan · zip — el original de la capa pull, que empareja Iterables por índice

waitAll, zip & friends

Lección

Demo 1 · Esperar a cada panel

Demo 2 · Por posición o por tiempo

Pruébalo tú

`waitAll`, `zip` & friends