`stopOn` & `startOn`

Two gates driven by a second stream: stopOn closes the chain when the trigger fires, startOn opens it.

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

Lecture

A push chain has no natural end. A pull pipeline stops when the consumer stops asking, but a live feed — a socket, a sensor, a Stream.periodic — keeps producing until something cancels it. Forgetting to do that is the classic leak: a screen is disposed, the widget is gone, and its subscription is still awake, still allocating, still calling setState on a corpse.

stopOn(trigger) makes the shutdown part of the chain rather than a variable you have to remember to cancel. The first event on trigger closes the output and cancels both subscriptions — the source's and the trigger's. Nothing the trigger carries is read, only that it fired, so its type is Stream<void> and any stream at all will do.

startOn(trigger) is the mirror: source events are dropped until the trigger fires once, and passed for good afterwards. It is the "wait until ready" gate — do not act on taps until the session has loaded. Note that it is unrelated to startWith, which prepends a value rather than gating the start; the two names sit close together and mean quite different things.

The …On suffix is the events layer's convention for "driven by a trigger stream", shared with sampleOn and chunkOn. fxdart events layer, after Rx's takeUntil and skipUntil — the names differ because Fx.takeUntil already means the predicate-driven takeUntilInclusive on the pull side, and one name cannot mean two things in one library.

Demo 1 · The off switch

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 · The on switch

Try it yourself

Exercise: a session window, built from both gates.

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

Related: FxSubscriptions — the other half of teardown: cancel many subscriptions at once · sampleOn — the same trigger convention, for reading instead of stopping · race — cancellation decided by whoever speaks first

stopOn & startOn

Lecture

Demo 1 · The off switch

Demo 2 · The on switch

Try it yourself

`stopOn` & `startOn`