`share` & `LiveValue.from`

Una ejecución de la cadena, muchos oyentes — y la versión que recuerda su último valor para quien llegue tarde.

FxEvents<T> FxEvents<T>.share({bool reset = true}) // chain (events); resubscribes after last-cancel unless complete LiveValue<T>.from(Stream<T> source) // hot, unseeded LiveValue<T>.seededFrom(T seed, Stream<T> source) // hot, seeded

Lección

Todos los operadores de esta sección construyen su propio StreamController, así que la cadena que devuelven es de suscripción única: escúchala dos veces y el segundo oyente recibe un StateError. Ese comportamiento por defecto es deliberado: mantiene la cadena fría, de modo que nada se ejecuta hasta que alguien la consume, y mantiene honesto el estado por oyente. Pero implica que dos widgets no pueden observar el mismo feed con debounce, throttle y switchMap sin construirlo dos veces.

share() lo arregla. Conecta con el primer oyente y difunde a todos los oyentes a partir de ahí, así que el trabajo aguas arriba ocurre una vez por muchos que estén mirando. Un temporizador de debounce, un socket, un map caro: uno de cada, no uno por suscriptor.

share({reset: true}) — el valor por defecto — ahora coincide con el reset por recuento de Rx. Cuando el último oyente se va antes de que la fuente haya completado, la suscripción de aguas arriba se cancela y el siguiente oyente empieza una suscripción nueva. Después de que la fuente completa, a un oyente tardío se le sigue entregando un stream cerrado. share(reset: false) es el comportamiento de 0.8.7: la última cancelación cierra para siempre. Una resuscripción necesita una fuente que permita un segundo listen — Stream.fromIterable, Stream.multi, FxEvents.defer, un broadcast — un StreamController de suscripción única ya gastado sigue sin poder volver a escucharse. Engancha todos los oyentes antes del primer evento si la fuente es de un solo disparo, o mantén uno vivo.

share() tampoco recuerda: un oyente que llega después de que haya pasado un evento simplemente se lo ha perdido. Para una ventana de historial, shareReplay es la página siguiente. Cuando los rezagados necesitan el estado actual —que es casi toda la UI—, LiveValue es la respuesta, y LiveValue.from(source) / LiveValue.seededFrom(seed, source) construyen uno directamente desde un stream. Esos son calientes: la suscripción se abre de inmediato, así que los valores que llegan antes de que nadie escuche siguen actualizando value, y close() cancela la fuente. Son constructores con nombre en vez de una semilla opcional para que un T nullable pueda sembrarse igualmente con null. Capa de eventos de fxdart, siguiendo a share y shareValue de Rx.

Demo 1 · Por qué un solo oyente es el comportamiento por defecto

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 {
  // Every FxEvents operator builds its own StreamController, so a chain
  // over a live source is SINGLE-SUBSCRIPTION: consumable exactly once.
  var runs = 0;
  final feed = StreamController<int>();
  final chain = fxEvents(feed.stream).map((v) {
    runs++;
    return v * 10;
  });

final consumer = chain.toList();

try {
    await chain.toList();
  } on StateError catch (e) {
    print('a second listener: $e');
  }

feed
    ..add(1)
    ..add(2)
    ..add(3);
  await feed.close();

print(await consumer); // [10, 20, 30]
  print('the chain ran $runs times');
  // Single-subscription is the right default — it keeps the chain cold
  // and cheap — but it means two widgets cannot watch the same feed.
}

Demo 2 · Una ejecución, dos oyentes

import 'dart:async';

import 'package:fxdart/fxdart.dart';

Future<void> main() async {
  var runs = 0;
  final shared = fxEvents(timed([(0, 1), (60, 2)], 150)).map((v) {
    runs++;
    return v * 10;
  }).share();

// Both listeners attach BEFORE the first event.
  final a = shared.toList();
  final b = shared.toList();

print(await a); // [10, 20]
  print(await b); // [10, 20]
  print('map ran $runs times for 2 listeners'); // 2, not 4

// After the source completes, a later listener is handed a closed
  // stream. reset: true (the default) only resubscribes when the last
  // listener left BEFORE complete — and only if the source allows a
  // second listen. share(reset: false) is the old forever-closed
  // behaviour even on a mid-run cancel.
  print(await shared.toList()); // [] — the source already completed

// fromIterable completes, so the second listen is empty even with
  // reset: true. A cancel-before-complete on a re-listenable source
  // would start a fresh run instead.
  final finished = fxEvents(Stream.fromIterable([7, 8])).share();
  print(await finished.toList()); // [7, 8]
  print(await finished.toList()); // []
}

Pruébalo tú

Ejercicio: un LiveValue alimentado directamente desde un stream.

import 'dart:async';

import 'package:fxdart/fxdart.dart';

Future<void> main() async {
  // Exercise: a LiveValue fed straight from a stream. Unlike share() it
  // remembers, so a late subscriber is not left with nothing.
  final sensor = StreamController<int>();
  final temperature = LiveValue.seededFrom(20, sensor.stream);

print(temperature.value); // 20 — the seed, before the source speaks

sensor.add(23);
  await Future<void>.delayed(Duration.zero);
  print(temperature.value); // 23 — updated with nobody listening yet

temperature.stream.listen((v) => print('late subscriber sees $v'));
  await Future<void>.delayed(Duration.zero); // replays 23 immediately

sensor.add(24);
  await Future<void>.delayed(Duration.zero); // then the live update

await temperature.close(); // cancels the source subscription too
  await sensor.close();
}

Relacionado: shareReplay — multicast que recuerda un búfer de historial · LiveValue — el compartir que recuerda: los suscriptores tardíos reciben primero el valor actual · tee — la respuesta del lado pull a dos lectores sobre una pasada, sin búfer · fork — dos cursores pull independientes sobre una fuente, a costa de un búfer

share & LiveValue.from

Lección

Demo 1 · Por qué un solo oyente es el comportamiento por defecto

Demo 2 · Una ejecución, dos oyentes

Pruébalo tú

`share` & `LiveValue.from`