`delay`, `spaceBy` & `sample`

Three ways to move events around in time: shift them all, spread them out, or read only the newest on a fixed clock.

FxEvents<T> FxEvents<T>.delay(Duration duration) // chain (events) FxEvents<T> FxEvents<T>.spaceBy(Duration gap) FxEvents<T> FxEvents<T>.sample(Duration period)

Lecture

Rate limiting always costs you something, and the only real question is what. throttle and debounce pay in events: they keep one per window and drop the rest, which is right when the events are samples of a continuous thing and an old one is worthless. spaceBy(gap) pays in time instead: every event survives, queued and released one per gap, which is right when each event is a discrete instruction you must not lose — six messages to send against an API that allows one call per 100ms.

That trade has a sharp edge. Because spaceBy queues rather than drops, a source that produces faster than gap forever grows an unbounded queue. It is for bursts — a batch that arrives at once and must all get through — not for genuinely endless input, where throttle's lossiness is a feature.

delay(duration) is the simplest of the three: the entire stream is shifted by a fixed amount, spacing intact, nothing dropped. The close waits for the last delayed event to land, so nothing is lost at the end; errors are forwarded immediately, since only data is worth holding.

sample(period) is sampleOn with the clock built in — the newest value every period, silent when nothing new has arrived. Reach for it when the source is a state-like feed (a position, a temperature, a scroll offset) and the consumer has its own refresh rate. fxdart events layer, after Rx's delay, interval and sampleTime.

Demo 1 · Pacing a burst, losslessly

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 {
  // Four requests queued up at once, against an API that allows one
  // call per 150ms.
  final watch = Stopwatch()..start();
  final sent = await fxEvents(Stream.fromIterable(['r1', 'r2', 'r3', 'r4']))
      .spaceBy(const Duration(milliseconds: 150))
      .toList();
  watch.stop();

print(sent); // [r1, r2, r3, r4] — every one of them
  print('took ~${(watch.elapsedMilliseconds / 100).round() * 100}ms');

// throttle, given the same burst, keeps ONE and drops the rest:
  final throttled = await fxEvents(Stream.fromIterable(['r1', 'r2', 'r3', 'r4']))
      .throttle(const Duration(milliseconds: 150))
      .toList();
  print(throttled); // [r1]
  // Same rate limit, opposite trade: spaceBy is lossless and slow,
  // throttle is lossy and immediate.
}

Demo 2 · Shifting, and reading on a clock

import 'dart:async';

import 'package:fxdart/fxdart.dart';

Future<void> main() async {
  // delay shifts the whole stream — nothing dropped, spacing preserved.
  final watch = Stopwatch()..start();
  final shifted = await fxEvents(Stream.fromIterable([1, 2, 3]))
      .delay(const Duration(milliseconds: 200))
      .toList();
  print('$shifted after ~${(watch.elapsedMilliseconds / 100).round() * 100}ms');

// sample reads the newest value on its own clock, like sampleOn but
  // with the clock built in.
  final sensor = timed([(0, 'a'), (60, 'b'), (300, 'c')], 640);
  final read = await fxEvents(sensor)
      .sample(const Duration(milliseconds: 200))
      .toList();

print(read); // [b, c]
  // The 200ms tick saw b, the 400ms tick saw c, and the 600ms tick had
  // nothing new to report — so it stayed silent.
}

Try it yourself

Exercise: a send rate and a reporting rate, in one chain.

import 'dart:async';

import 'package:fxdart/fxdart.dart';

Future<void> main() async {
  // Exercise: a burst of outbound messages, paced to the wire and
  // then batched for the log.
  final outbox = Stream.fromIterable(['m1', 'm2', 'm3', 'm4', 'm5', 'm6']);

final logged = await fxEvents(outbox)
      .spaceBy(const Duration(milliseconds: 100))
      .chunkEvery(const Duration(milliseconds: 250))
      .toList();

print(logged); // [[m1, m2], [m3, m4], [m5, m6]] (roughly)
  // spaceBy sets the send rate; chunkEvery sets the reporting rate.
  // Because spaceBy queues rather than drops, an unbounded burst grows
  // an unbounded queue — reach for throttle or debounce when the input
  // is genuinely endless.
}

Related: throttle — the lossy counterpart: one event per window, immediately · debounce — wait for the burst to end, then take its last value · chunkEvery — keep every event too, but grouped rather than spread out

delay, spaceBy & sample

Lecture

Demo 1 · Pacing a burst, losslessly

Demo 2 · Shifting, and reading on a clock

Try it yourself

`delay`, `spaceBy` & `sample`