`stopOn` & `startOn`

Dos compuertas gobernadas por un segundo stream: stopOn cierra la cadena cuando el disparador salta, startOn la abre.

FxEvents<T> FxEvents<T>.stopOn(Stream<void> trigger) // chain (events) FxEvents<T> FxEvents<T>.startOn(Stream<void> trigger) // chain (events)

Lección

Una cadena push no tiene un final natural. Un pipeline pull para cuando el consumidor deja de pedir, pero un feed en vivo — un socket, un sensor, un Stream.periodic — sigue produciendo hasta que algo lo cancela. Olvidarse de hacerlo es la fuga clásica: se destruye una pantalla, el widget ya no está, y su suscripción sigue despierta, sigue reservando memoria, sigue llamando a setState sobre un cadáver.

stopOn(trigger) convierte el apagado en parte de la cadena en vez de en una variable que tienes que acordarte de cancelar. El primer evento de trigger cierra la salida y cancela las dos suscripciones: la de la fuente y la del disparador. No se lee nada de lo que lleve el disparador, solo que disparó, así que su tipo es Stream<void> y sirve cualquier stream.

startOn(trigger) es el espejo: los eventos de la fuente se descartan hasta que el disparador salta una vez, y a partir de ahí pasan para siempre. Es la compuerta de «espera a estar listo»: no actúes sobre los toques hasta que la sesión haya cargado. Ojo, no tiene relación con startWith, que antepone un valor en lugar de controlar el arranque; los dos nombres quedan cerca y significan cosas bastante distintas.

El sufijo …On es el convenio de la capa de eventos para «gobernado por un stream disparador», compartido con sampleOn y chunkOn. Capa de eventos de fxdart, siguiendo a takeUntil y skipUntil de Rx: los nombres difieren porque Fx.takeUntil ya significa el takeUntilInclusive gobernado por predicado del lado pull, y un nombre no puede significar dos cosas en una misma biblioteca.

Demo 1 · El interruptor de apagado

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 {
  // A telemetry feed with no natural end — it would run for 5 seconds.
  final telemetry = timed([
    (0, 'cpu 12%'),
    (60, 'cpu 31%'),
    (120, 'cpu 44%'),
    (300, 'cpu 51%'),
  ], 5000);

// The screen is torn down at 200ms.
  final disposed = timed([(200, null)], 240);

final watch = Stopwatch()..start();
  final seen = await fxEvents(telemetry).stopOn(disposed).toList();
  watch.stop();

print(seen); // [cpu 12%, cpu 31%, cpu 44%]
  print('closed after ~${(watch.elapsedMilliseconds / 100).round() * 100}ms');
  // stopOn did not wait out the source's 5 seconds: the first trigger
  // event closed the chain and cancelled BOTH subscriptions.
}

Demo 2 · El interruptor de encendido

Pruébalo tú

Ejercicio: una ventana de sesión, construida con las dos compuertas.

import 'dart:async';

import 'package:fxdart/fxdart.dart';

Future<void> main() async {
  // Exercise: record only the ticks that fall inside a session.
  final ticks = timed([
    (0, 1),
    (150, 2),
    (300, 3),
    (450, 4),
    (600, 5),
    (750, 6),
    (900, 7),
    (1050, 8),
  ], 1200);

final loggedIn = timed([(375, null)], 1300);
  final loggedOut = timed([(825, null)], 1300);

final session = await fxEvents(ticks)
      .startOn(loggedIn)
      .stopOn(loggedOut)
      .toList();

print(session); // [4, 5, 6]
  // The two gates compose: startOn opens at 375ms, stopOn closes at
  // 825ms, and the ticks outside that window never reach the chain.
}

Relacionado: FxSubscriptions — la otra mitad del desmontaje: cancelar muchas suscripciones a la vez · sampleOn — el mismo convenio de disparador, para leer en vez de parar · race — cancelación decidida por quien hable primero

stopOn & startOn

Lección

Demo 1 · El interruptor de apagado

Demo 2 · El interruptor de encendido

Pruébalo tú

`stopOn` & `startOn`