`windowOn`, `windowCount` & `windowEvery`

Streams vivos anidados: ve los valores de una ventana antes de que cierre, rota por cantidad, por un disparador o por un reloj.

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)

Lección

chunk espera a que una ventana cierre y entonces emite una List. La familia window* emite la ventana mientras sigue abierta: cada valor del stream externo es un FxEvents anidado, así que un suscriptor puede ver eventos antes del cierre — un gráfico en vivo del minuto actual, un total acumulado del lote actual. Esa es toda la diferencia, y es por lo que el tipo de retorno es FxEvents<FxEvents<T>>.

windowCount(size) rota cada size eventos; un startEvery menor que size solapa, uno mayor deja huecos. windowOn(boundaries) abre una ventana de inmediato al escuchar y rota en cada valor disparador — se ignora la finalización del límite, así que la ventana actual sigue abierta hasta que la fuente completa. windowEvery(span) es la forma reloj; every solapa o deja huecos en ese periodo, y maxSize cierra una ventana antes por cantidad.

La vida útil sigue a RxJS 9: cancelar el externo completa los internos vivos en silencio en vez de darles error, así que los suscriptores anidados se desmontan limpiamente. Un error de la fuente sigue dando error a cada interno vivo, y después al externo. Y como los internos son streams, puede aparecer una ventana vacía al final cuando se abre una nueva al llenar la anterior el último valor — chunk* se salta esas; window* no.

chunkToggle(openings, closeOf) es la contraparte de la familia lista de windowToggle: cada apertura arranca un búfer, el primer evento de closeOf de esa apertura lo emite, los búferes vacíos se saltan como chunkOn. Capa de eventos de fxdart, siguiendo a window, windowCount, windowTime y bufferToggle de Rx.

Demo 1 · Ventanas de dos

Demo 2 · Rotar con un disparador

Pruébalo tú

Ejercicio: internos vivos que abarcan un Duration corto.

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

Relacionado: chunk / chunkOn — las mismas ventanas como listas, emitidas cuando cierran · windowed — listas deslizantes en la capa pull · groupsBy — internos vivos con clave por valor, no por tiempo

windowOn, windowCount & windowEvery

Lección

Demo 1 · Ventanas de dos

Demo 2 · Rotar con un disparador

Pruébalo tú

`windowOn`, `windowCount` & `windowEvery`