`fxEvents`

Wraps a plain Dart Stream in a chainable FxEvents — the entry point to FxDart's push side.

FxEvents<T> fxEvents<T>(Stream<T> stream) class FxEvents<T> { Stream<T> get stream; // unwrap FxEvents<R> map<R>(R Function(T a) f); FxEvents<T> where(bool Function(T a) f); FxEvents<R> asyncMap<R>(FutureOr<R> Function(T a) f); FxEvents<T> startWith(T value); FxEvents<R> scan<R>(R Function(R acc, T a) f, R seed); // seed first FxEvents<T> uniqAdjacent() / uniqAdjacentBy<B>(B Function(T a) f); FxEvents<(T, T)> pairwise(); FxEvents<T> take(int count) / drop(int count) / skip(int count); static FxEvents<T> merge<T>(Iterable<Stream<T>> sources); Future<List<T>> toList(); // terminal Future<T?> head() / firstOrNull(); // terminal FxAsync<T> pull(); // cross into the pull model }

Lecture

Everything before this section is pull: a pipeline sits still until a terminal operator demands the next item. But some problems are genuinely push — keystrokes, sensor readings, socket messages arrive when they arrive, whether anyone asked or not. Those are what Dart's Stream models, and fxEvents(stream) gives that world the same chainable treatment: map, where, asyncMap, startWith, FxEvents.merge — plus the time and combination operators the rest of this section covers.

Design decisions worth knowing. FxEvents is a thin wrapper, deliberately not a set of Stream extensions — so its operators can never collide with rxdart or any other stream library in the same file. The one exception is the .fxEvents entry getter, a single name nothing else claims; it sits beside .fx in Stream bridges. The chain stays cold: wrapping listens to nothing; only a terminal (toList, head, listen) starts events flowing. And it is an fxdart extension inspired by Rx, not part of FxTS — the ideas come from Rx, but where a name would clash with the pull layer's the pull spelling wins: uniqAdjacent rather than distinctUntilChanged, stopOn rather than takeUntil, head rather than first. One word means one thing on both sides.

Two escape hatches keep you unlocked. .stream unwraps back to a plain Stream for any Stream-based API, at any point in the chain. And .pull() crosses into the typed pull world: the events become an FxAsync chain, pulled on demand from there on — push at the edge where events are born, pull in the core where you control demand.

Demo 1 · A cold chain over a Stream

Demo 2 · merge, and crossing into the pull world

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 {
  // Two sensors, one feed: merge interleaves in arrival order and closes
  // once every source has closed.
  final merged = await FxEvents.merge([
    timed([(0, 'kitchen 21.0'), (100, 'kitchen 21.5')], 200),
    timed([(50, 'attic 15.2'), (150, 'attic 15.1')], 220),
  ]).toList();
  print(merged); // [kitchen 21.0, attic 15.2, kitchen 21.5, attic 15.1]

// .pull() crosses into the typed pull world: from here on the events
  // are an FxAsync chain, pulled on demand.
  final odds = await fxEvents(Stream.fromIterable([1, 2, 3, 4, 5]))
      .pull()
      .filter((v) => v.isOdd)
      .map((v) => v * 2)
      .toList();
  print(odds); // [2, 6, 10]
}

Try it yourself

Exercise: clean up a glitchy sensor feed.

Related: Stream bridges — the pull side of the border, and stream.fx vs stream.fxEvents side by side · debounce & throttle — both have FxEvents forms · LiveValue — the current-value companion to this chain