`materialize`, `timestamped`, `sequenceEqual`

Reify notifications as Next / Err / Done, stamp events with time, and test two sequences for equality.

sealed class StreamEvent<T> — Next(T value) | Err(Object error, [StackTrace]) | Done() FxEvents<StreamEvent<T>> FxEvents<T>.materialize() // chain (events) FxEvents<T> FxEvents<StreamEvent<T>>.dematerialize() FxEvents<(DateTime at, T value)> FxEvents<T>.timestamped({DateTime Function()? now}) FxEvents<(Duration dt, T value)> FxEvents<T>.intervals({DateTime Function()? now}) (FxEvents<T> matches, FxEvents<T> rest) FxEvents<T>.partition(bool Function(T) test) Future<bool> FxEvents<T>.sequenceEqual(Stream<T> other, {bool Function(T, T)? eq}) bool sequenceEqual<T>(Iterable<T> a, Iterable<T> b, [bool Function(T, T)? eq]) Future<bool> sequenceEqualAsync<T>(FxAsyncIterable<T> a, FxAsyncIterable<T> b, [bool Function(T, T)? eq]) bool Fx.sequenceEqual(Iterable<T> other, [bool Function(T, T)? eq]) Future<bool> FxAsync.sequenceEqual(FxAsyncIterable<T> other, [bool Function(T, T)? eq])

Lecture

A Dart Stream's three terminals — a value, an error, a close — normally leave the pipe. materialize turns each into a StreamEvent value that can travel through the chain: a data event becomes Next(value), an error becomes Err and then the result completes — it does not error — and a close becomes Done and then the result completes. That is the point of reifying them: toList can collect an error instead of failing, a log can print Err(boom) next to Next(1), a test can assert the exact notification sequence. dematerialize is the inverse — Next becomes a value, Err becomes Stream.addError, Done closes the result, and anything after Done is ignored. Round-trip: materialize().dematerialize() is the original stream of values. fxdart events layer, after Rx's materialize / dematerialize.

Time is the other metadata a notification can carry. timestamped pairs each event with the wall-clock time it arrived ((DateTime at, T value)); intervals pairs it with the time since the previous one ((Duration dt, T value)), and the first event is always Duration.zero. Both take now: so a test can pass a fake clock — now: () => DateTime.utc(2020) — instead of DateTime.now. Errors and close pass through unchanged. After Rx's timestamp and timeInterval.

partition(test) on events is not the pull-side partition (which walks once and returns two lists). It splits one live chain into (matches, rest) sharing one run of the source. The record is returned eagerly; listening to either side starts the source; a value that belongs to a side nobody is listening to is dropped, not buffered. Listen to both before the source fires if you want both halves.

sequenceEqual asks whether two sequences hold the same values in the same order and stop together. On the events layer it is a terminal: fxEvents(a).sequenceEqual(b) returns Future<bool>, false on the first value or length mismatch, and an error from either side fails the future. The same question exists on pull: sequenceEqual / Fx.sequenceEqual for iterables, sequenceEqualAsync / FxAsync.sequenceEqual for FxAsyncIterables. After Rx's sequenceEqual.

Demo 1 · Next, Err, Done

A clean close becomes Done. An error becomes Err and then the chain completes — so toList returns the StreamEvent list instead of failing:

Demo 2 · A clock you can pass in

now: () => DateTime.utc(2020) keeps the stamps deterministic. intervals uses the same hook, with a stepping clock so the gaps are exact:

Demo 3 · sequenceEqual, and partition

The pull spelling is just fx([1, 2]).sequenceEqual([1, 2]). The events terminal takes a Stream. Listen to both sides of partition before the source runs, or the un-listened half is dropped:

import 'package:fxdart/fxdart.dart';

Future<void> main() async {
  // Pull / sync — fx([1, 2]).sequenceEqual([1, 2])
  print(fx([1, 2]).sequenceEqual([1, 2])); // true
  print(fx([1, 2]).sequenceEqual([1, 9])); // false
  print(fx([1, 2]).sequenceEqual([1])); // false — length mismatch

// Pull / async
  print(await fx([1, 2]).toAsync().sequenceEqual(toAsync([1, 2]))); // true

// Push / events — same values, same order, complete together.
  print(
    await fxEvents(
      Stream.fromIterable([1, 2]),
    ).sequenceEqual(Stream.fromIterable([1, 2])),
  ); // true
  print(
    await fxEvents(
      Stream.fromIterable([1, 2]),
    ).sequenceEqual(Stream.fromIterable([1, 9])),
  ); // false

// Events partition: one source run, two chains. Listen to both
  // before the source fires — a side nobody is listening to is dropped.
  final (evens, odds) = fxEvents(
    Stream.fromIterable([1, 2, 3, 4]),
  ).partition((n) => n.isEven);
  final evenF = evens.toList();
  final oddF = odds.toList();
  print(await evenF); // [2, 4]
  print(await oddF); // [1, 3]
}

Related: fxEvents — the chain these operators sit on · Stream bridges — four ways to pull a Stream into FxAsync · partition — the pull-side original, two lists from one walk · FxSubscriptions — cancel a bag of listeners together

materialize, timestamped, sequenceEqual

Lecture

Demo 1 · Next, Err, Done

Demo 2 · A clock you can pass in

Demo 3 · sequenceEqual, and partition

`materialize`, `timestamped`, `sequenceEqual`