`chunk`, `chunkOn` & `chunkEvery`

Recoge eventos en listas — por cantidad, por disparador o por reloj — para que un stream parlanchín se convierta en unas pocas llamadas agrupadas.

FxEvents<List<T>> FxEvents<T>.chunk(int count) // chain (events) FxEvents<List<T>> FxEvents<T>.chunkOn(Stream<void> trigger) FxEvents<List<T>> FxEvents<T>.chunkEvery(Duration window)

Lección

Agrupar es la mejora de rendimiento más barata que tiene un stream parlanchín. Cien eventos de analítica son cien viajes de ida y vuelta uno a uno, y dos viajes en lotes de cincuenta; el trabajo es idéntico, el coste no. La capa pull agrupa por cantidad con chunk, y eso es todo lo que puede hacer: un pipeline pull no tiene reloj, así que «todo lo que ha pasado en los últimos dos segundos» no es una pregunta que pueda formular.

El lado push sí puede. chunk(count) es esa misma agrupación de tamaño fijo, con un último lote corto que se vacía cuando la fuente se cierra para que no quede nada varado. chunkEvery(window) agrupa por tiempo en su lugar: lo que haya llegado en la última ventana se emite como una lista. Y chunkOn(trigger) entrega la decisión a un segundo stream: agrupa cuando el usuario hace scroll, cuando termina el frame, cuando vuelve la conexión.

Las dos formas gobernadas por el tiempo callan ante una ventana vacía. Un tic que no encuentra nada acumulado no emite nada en lugar de una lista vacía, así que el código de aguas abajo nunca tiene que filtrar lotes que significan «no ha pasado nada»: la misma regla de honestidad que sigue sampleOn. Lo que siga acumulado cuando la fuente se cierre se vacía antes del cierre.

Capa de eventos de fxdart, siguiendo a bufferCount, buffer y bufferTime de Rx. La familia conserva el chunk de la capa pull como palabra raíz —un nombre para una idea en las dos mitades de la biblioteca— con el sufijo …On para un disparador y …Every para un reloj.

Demo 1 · Agrupar por reloj

Demo 2 · Por cantidad y por disparador

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 {
  // By count: fixed-size batches, with a short final one.
  final byCount =
      await fxEvents(Stream.fromIterable([1, 2, 3, 4, 5])).chunk(2).toList();
  print(byCount); // [[1, 2], [3, 4], [5]]

// By trigger: someone else decides when a batch is due.
  final rows = timed([(0, 'a'), (40, 'b'), (200, 'c')], 400);
  final flush = timed([(100, null), (150, null), (300, null)], 420);

final byTrigger = await fxEvents(rows).chunkOn(flush).toList();
  print(byTrigger); // [[a, b], [c]]
  // The 100ms flush took [a, b]; the 150ms flush found an empty buffer
  // and stayed silent; the 300ms flush took [c].
}

Pruébalo tú

Ejercicio: un resumen por ventana en vez de un informe por clic.

import 'dart:async';

import 'package:fxdart/fxdart.dart';

Future<void> main() async {
  // Exercise: turn a noisy click stream into a per-window summary
  // instead of one report per click.
  final clicks = timed([
    (0, 'home'),
    (40, 'home'),
    (80, 'search'),
    (250, 'cart'),
    (290, 'cart'),
    (330, 'checkout'),
  ], 620);

final report = await fxEvents(clicks)
      .chunkEvery(const Duration(milliseconds: 200))
      .map((batch) => '${batch.length} clicks: ${batch.toSet().join(", ")}')
      .toList();

report.forEach(print);
  // 3 clicks: home, search
  // 3 clicks: cart, checkout
  //
  // chunk* is the push-side counterpart of the pull layer's chunk: same
  // idea, but a window can be closed by a clock or a trigger, which a
  // pull pipeline has no way to express.
}

Relacionado: chunk — el original de la capa pull, que agrupa por cantidad sobre un Iterable · throttle — cuando quieres un evento por ventana en vez de todos · spaceBy — la otra forma de frenar una ráfaga: estirarla en vez de agruparla

chunk, chunkOn & chunkEvery

Lección

Demo 1 · Agrupar por reloj

Demo 2 · Por cantidad y por disparador

Pruébalo tú

`chunk`, `chunkOn` & `chunkEvery`