`FxSubscriptions`

Una bolsa de suscripciones, canceladas a la vez — para que el desmontaje sea una llamada en vez de un campo por 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(); }

Lección

Un objeto que escucha varios streams tiene que mantener viva cada suscripción por un único motivo: volver a cancelarla más tarde. El resultado es el montón conocido de campos nullables, cada uno declarado arriba, cada uno asignado en initState, cada uno cancelado en dispose; y la fuga siempre es la que alguien se olvidó de añadir a la tercera lista.

FxSubscriptions reduce eso a un solo objeto. add mete una suscripción en la bolsa y la devuelve, así que se lee como una expresión y no como una sentencia, y cancelAll() las termina todas. El desmontaje se convierte en una sola línea: Future<void> dispose() => subs.cancelAll();

pauseAll() y resumeAll() son la versión suave, para cuando el trabajo debe pararse sin que se deshaga el cableado: una pantalla que pasa a segundo plano, una pestaña que pierde el foco. Las suscripciones en pausa almacenan en vez de descartar, así que no se pierde nada durante el hueco.

La bolsa se vacía antes de esperar sus cancelaciones, así que un segundo cancelAll() durante la espera no puede cancelar nada dos veces, y el mismo objeto puede albergar después una generación nueva de suscripciones. Capa de eventos de fxdart, siguiendo a CompositeSubscription de Rx.

Se lleva de forma natural con stopOn: usa stopOn cuando una cadena debe terminar porque algo ha pasado, y FxSubscriptions cuando un conjunto de cadenas debe terminar porque lo que las poseía se va.

Demo 1 · El dispose de una línea

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 · Pausar sin desmontar

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();
}

Pruébalo tú

Ejercicio: addAll, y reutilizar la bolsa después de un 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();
}

Relacionado: stopOn — desmontaje gobernado por un evento en vez de por el ciclo de vida de un propietario · fxEvents — la cadena cuyo listen te entrega las suscripciones que esto guarda · LiveValue — tiene su propio close(), y esta bolsa no lo guarda