`windowOn`, `windowCount` & `windowEvery`

Nested live streams: see the values of a window before it closes, rotate by count, by a trigger, or by a clock.

FxEvents<FxEvents<T>> FxEvents<T>.windowOn(Stream<void> boundaries) // chain (events) FxEvents<FxEvents<T>> FxEvents<T>.windowCount(int size, {int? startEvery}) FxEvents<FxEvents<T>> FxEvents<T>.windowEvery(Duration span, {Duration? every, int? maxSize}) FxEvents<List<T>> FxEvents<T>.chunkToggle<O>(Stream<O> openings, Stream<void> Function(O opening) closeOf)

Lecture

chunk waits for a window to close and then emits a List. The window* family emits the window while it is still open: each value of the outer stream is a nested FxEvents, so a subscriber can see events before the close — a live chart of the current minute, a running total of the current batch. That is the whole difference, and it is why the return type is FxEvents<FxEvents<T>>.

windowCount(size) rotates every size events; startEvery smaller than size overlaps, larger gaps. windowOn(boundaries) opens a window immediately on listen and rotates on each trigger value — boundary completion is ignored, so the current window stays open until the source completes. windowEvery(span) is the clock form; every overlaps or gaps on that period, and maxSize closes a window early by count.

Lifetime follows RxJS 9: cancelling the outer completes live inners silently rather than erroring them, so nested subscribers tear down cleanly. A source error still errors every live inner, then the outer. And because inners are streams, a trailing empty window can appear when a new one opens as the last value fills the previous — chunk* skips those; window* does not.

chunkToggle(openings, closeOf) is the list-family counterpart of windowToggle: each opening starts a buffer, the first event from closeOf of that opening emits it, empty buffers are skipped like chunkOn. fxdart events layer, after Rx's window, windowCount, windowTime, and bufferToggle.

Demo 1 · Windows of two

Demo 2 · Rotate on a trigger

Try it yourself

Exercise: live inners spanning a short Duration.

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 {
  // Exercise: windowEvery is the clock form — live inners spanning a
  // Duration, tumbling when every is omitted. chunkToggle is the
  // list-family counterpart of windowToggle: overlapping buffers,
  // empty skipped, like chunkOn.
  final events = timed([(0, 1), (30, 2), (60, 3), (150, 4)], 180);

final windows = await fxEvents(events)
      .windowEvery(const Duration(milliseconds: 80))
      .toList();

for (final w in windows) {
    final list = await w.toList();
    if (list.isNotEmpty) print(list);
  }
  // [1, 2, 3]
  // [4]
  //
  // The first 80ms window took 1, 2, 3; the next took 4. every: and
  // maxSize close or overlap windows early — the same knobs as
  // windowCount's startEvery, on a clock.
}

Related: chunk / chunkOn — the same windows as lists, emitted when they close · windowed — sliding lists on the pull layer · groupsBy — live inners keyed by value, not by time

windowOn, windowCount & windowEvery

Lecture

Demo 1 · Windows of two

Demo 2 · Rotate on a trigger

Try it yourself

`windowOn`, `windowCount` & `windowEvery`