`delay`, `spaceBy` & `sample`

Tres formas de mover eventos en el tiempo: desplazarlos todos, separarlos, o leer solo el más reciente a un ritmo fijo.

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

Lección

Limitar el ritmo siempre te cuesta algo, y la única pregunta real es qué. throttle y debounce pagan en eventos: se quedan con uno por ventana y descartan el resto, lo cual es correcto cuando los eventos son muestras de algo continuo y uno viejo no vale nada. spaceBy(gap) paga en tiempo: todos los eventos sobreviven, encolados y liberados uno por gap, lo cual es correcto cuando cada evento es una instrucción discreta que no puedes perder — seis mensajes que enviar contra una API que permite una llamada cada 100 ms.

Ese intercambio tiene un filo. Como spaceBy encola en vez de descartar, una fuente que produzca más rápido que gap indefinidamente hace crecer una cola sin límite. Es para ráfagas —un lote que llega de golpe y tiene que pasar entero—, no para una entrada genuinamente interminable, donde perder eventos es justamente la virtud de throttle.

delay(duration) es el más simple de los tres: el stream entero se desplaza una cantidad fija, con su separación intacta y sin descartar nada. El cierre espera a que aterrice el último evento retrasado, así que no se pierde nada al final; los errores se reenvían de inmediato, ya que solo los datos merecen retenerse.

sample(period) es sampleOn con el reloj incorporado: el valor más reciente cada period, en silencio cuando no ha llegado nada nuevo. Échale mano cuando la fuente sea un feed con aire de estado (una posición, una temperatura, un desplazamiento de scroll) y el consumidor tenga su propia tasa de refresco. Capa de eventos de fxdart, siguiendo a delay, interval y sampleTime de Rx.

Demo 1 · Dosificar una ráfaga sin pérdidas

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 · Desplazar, y leer por reloj

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

Pruébalo tú

Ejercicio: un ritmo de envío y un ritmo de informe, en una sola cadena.

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

Relacionado: throttle — la contraparte con pérdidas: un evento por ventana, de inmediato · debounce — espera a que acabe la ráfaga y toma su último valor · chunkEvery — también conserva todos los eventos, pero agrupados en vez de separados

delay, spaceBy & sample

Lección

Demo 1 · Dosificar una ráfaga sin pérdidas

Demo 2 · Desplazar, y leer por reloj

Pruébalo tú

`delay`, `spaceBy` & `sample`