`combine` & `CombineSpec`

One combinator for "who triggers" and "who must have spoken" — plus zipAll and withLatestFromAll.

FxEvents<List<T>> combine<T>(Iterable<CombineSpec<T>> specs) class CombineSpec<T>(Stream<T> source, {bool causesEmit = true, bool requireFirst = true}) FxEvents<R> FxEvents<Stream<T>>.zipAll<R>([R Function(List<T> values)? project]) FxEvents<R> FxEvents<T>.withLatestFromAll<R>(Iterable<Stream<Object?>> others, R Function(T value, List<Object?> latest) combine)

Lecture

combineLatest, withLatestFrom and zip differ in who may fire an emit and whether every side must have spoken. combine is the unified form: a list of CombineSpec, each a source plus two flags. It is a top-level function, not FxEvents.combine — Dart cannot add statics to FxEvents from another file.

causesEmit: true (the default) means an event from this source may produce an output, once every requireFirst spec has a value. All-true specs are FxEvents.combineLatestAll. causesEmit: false means this source is context only: it updates the slot but never fires an emit on its own — that is withLatestFrom. requireFirst: true (the default) withholds every emit until this source has produced at least one value; false lets the slot be null until it speaks. The result closes when every source has closed.

Two more combinators live next to it. zipAll on FxEvents<Stream<T>> collects every inner stream until the outer completes, then zips them by index — inners are not even subscribed until then. withLatestFromAll is the N-ary withLatestFrom: on every source event, combine it with the latest of every other, dropping source events that arrive before every other has spoken. fxdart events layer, after Rx's combineLatest / withLatestFrom / zipAll.

Demo 1 · Both sides trigger

Demo 2 · Context only — causesEmit: false

Try it yourself

Exercise: zipAll on a stream of streams, and withLatestFromAll on a source.

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 {
  // zipAll waits for the OUTER to complete, then zips collected inners
  // by index. Write zipAll<List<int>>() when omitting project.
  final zipped = await fxEvents(Stream.fromIterable([
    Stream.fromIterable([1, 2]),
    Stream.fromIterable([10, 20]),
  ])).zipAll<List<int>>().toList();
  print(zipped); // [[1, 10], [2, 20]]

// withLatestFromAll is N-ary withLatestFrom: only SOURCE events emit,
  // stamped with every other's latest. Early source events are dropped
  // until every other has spoken.
  final stamped = await fxEvents(timed([(0, 'r1'), (80, 'r2'), (160, 'r3')], 220))
      .withLatestFromAll(
        [timed([(40, 'v1'), (120, 'v2')], 220)],
        (v, latest) => '$v@${latest.first}',
      )
      .toList();
  print(stamped); // [r2@v1, r3@v2]
}

Related: combineLatest — both sides trigger, the all-true CombineSpec · withLatestFrom — one-sided, the causesEmit: false spec · waitAll — zip, combineLatestAll, concat

combine & CombineSpec

Lecture

Demo 1 · Both sides trigger

Demo 2 · Context only — causesEmit: false

Try it yourself

`combine` & `CombineSpec`