`chunk`, `chunkOn` & `chunkEvery`

Collect events into lists — by count, by a trigger, or by a clock — so a chatty stream becomes a few batched calls.

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)

Lecture

Batching is the cheapest performance win a chatty stream has. A hundred analytics events are a hundred round trips one at a time and two round trips in batches of fifty; the work is identical, the cost is not. The pull layer batches by count with chunk, and that is all it can do — a pull pipeline has no clock, so "everything that happened in the last two seconds" is not a question it can ask.

The push side can. chunk(count) is the same fixed-size batching, with a short final batch flushed when the source closes so nothing is stranded. chunkEvery(window) batches by time instead: whatever arrived in the last window is emitted as one list. And chunkOn(trigger) hands the decision to a second stream — batch when the user scrolls, when the frame ends, when the connection comes back.

Both time-driven forms are silent on an empty window. A tick that finds nothing buffered emits nothing rather than an empty list, so downstream code never has to filter out batches that mean "nothing happened" — the same honesty rule sampleOn follows. Whatever is still buffered when the source closes is flushed before the close.

fxdart events layer, after Rx's bufferCount, buffer and bufferTime. The family keeps the pull layer's chunk as its root word — one name for one idea across both halves of the library — with the …On suffix for a trigger and …Every for a clock.

Demo 1 · Batching by the clock

Demo 2 · By count, and by trigger

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

Try it yourself

Exercise: a per-window summary instead of one report per click.

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

Related: chunk — the pull-layer original, batching by count over an Iterable · throttle — when you want one event per window rather than all of them · spaceBy — the other way to slow a burst: stretch it instead of grouping it

chunk, chunkOn & chunkEvery

Lecture

Demo 1 · Batching by the clock

Demo 2 · By count, and by trigger

Try it yourself

`chunk`, `chunkOn` & `chunkEvery`