`FxSubscriptions`

A bag of subscriptions, cancelled together — so teardown is one call instead of one field per stream.

class FxSubscriptions { int get length; bool get isEmpty; bool get isNotEmpty; StreamSubscription<T> add<T>(StreamSubscription<T> subscription); void addAll(Iterable<StreamSubscription<void>> subscriptions); Future<void> cancelAll(); // cancel every one, and empty the bag void pauseAll(); void resumeAll(); }

Lecture

An object that listens to several streams has to keep every subscription alive for one reason only: to cancel it again later. The result is the familiar pile of nullable fields, each declared at the top, each assigned in initState, each cancelled in dispose — and the leak is always the one somebody forgot to add to the third list.

FxSubscriptions collapses that to one object. add puts a subscription in the bag and returns it, so it reads as an expression rather than a statement, and cancelAll() ends every one of them. Teardown becomes a single line: Future<void> dispose() => subs.cancelAll();

pauseAll() and resumeAll() are the softer version, for when the work should stop without the wiring coming apart — a screen going to the background, a tab losing focus. Paused subscriptions buffer rather than drop, so nothing is lost across the gap.

The bag is emptied before its cancellations are awaited, so a second cancelAll() during the wait cannot cancel anything twice, and the same object can hold a fresh generation of subscriptions afterwards. fxdart events layer, after Rx's CompositeSubscription.

It pairs naturally with stopOn: use stopOn when a chain should end because something happened, and FxSubscriptions when a set of chains should end because the thing that owned them is going away.

Demo 1 · The dispose one-liner

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 screen listening to three feeds at once.
  final ticks = StreamController<int>();
  final clicks = StreamController<String>();

final subs = FxSubscriptions();
  subs
    ..add(fxEvents(ticks.stream).map((v) => v * 2).listen((v) => print('tick $v')))
    ..add(fxEvents(clicks.stream).listen((v) => print('click $v')));

print('holding ${subs.length} subscriptions');

ticks.add(1);
  clicks.add('home');
  await Future<void>.delayed(Duration.zero);

// One call ends them all — the dispose() one-liner.
  await subs.cancelAll();
  print('after cancelAll: isEmpty = ${subs.isEmpty}');

ticks.add(2);
  clicks.add('cart');
  await Future<void>.delayed(Duration.zero); // nothing prints

await ticks.close();
  await clicks.close();
}

Demo 2 · Pausing without tearing down

import 'dart:async';

import 'package:fxdart/fxdart.dart';

Future<void> main() async {
  final a = StreamController<int>();
  final b = StreamController<int>();
  final seen = <int>[];

final subs = FxSubscriptions();
  subs
    ..add(a.stream.listen(seen.add))
    ..add(b.stream.listen(seen.add));

// Backgrounded: hold everything without tearing it down.
  subs.pauseAll();
  a.add(1);
  b.add(2);
  await Future<void>.delayed(Duration.zero);
  print(seen); // [] — buffered, not lost

// Foregrounded again.
  subs.resumeAll();
  await Future<void>.delayed(Duration.zero);
  print(seen); // [1, 2]

await subs.cancelAll();
  await a.close();
  await b.close();
}

Try it yourself

Exercise: addAll, and reusing the bag after a cancel.

import 'dart:async';

import 'package:fxdart/fxdart.dart';

Future<void> main() async {
  // Exercise: the bag is reusable. cancelAll empties it, so the same
  // object can hold a fresh generation of subscriptions afterwards.
  final feed = StreamController<int>.broadcast();
  final subs = FxSubscriptions();

// add returns what it was handed, so it reads as an expression.
  final first = subs.add(feed.stream.listen((v) => print('gen 1: $v')));
  print('same subscription back: ${identical(first, first)}');

feed.add(1);
  await Future<void>.delayed(Duration.zero);

await subs.cancelAll();

// addAll takes a batch — a second generation, after the first is gone.
  subs.addAll([
    feed.stream.listen((v) => print('gen 2a: $v')),
    feed.stream.listen((v) => print('gen 2b: $v')),
  ]);
  print('holding ${subs.length} again');

feed.add(2);
  await Future<void>.delayed(Duration.zero);

await subs.cancelAll();
  await feed.close();
}

Related: stopOn — teardown driven by an event rather than by an owner's lifecycle · fxEvents — the chain whose listen hands you the subscriptions this holds · LiveValue — has its own close(), and is not held by this bag