`share` & `LiveValue.from`

One run of a chain, many listeners — and the version that remembers its latest value for whoever arrives late.

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

Lecture

Every operator in this section builds its own StreamController, so the chain it returns is single-subscription: listen to it twice and the second listener gets a StateError. That default is deliberate — it keeps the chain cold, so nothing runs until someone consumes it, and it keeps per-listener state honest. But it means two widgets cannot watch the same debounced, throttled, switch-mapped feed without building it twice.

share() fixes that. It connects on the first listener and broadcasts to every listener from there, so the work upstream happens once no matter how many are watching. A debounce timer, a socket, an expensive map — one of each, not one per subscriber.

share({reset: true}) — the default — now matches Rx's ref-count reset. When the last listener leaves before the source has completed, the upstream subscription is cancelled and the next listener starts a fresh subscribe. After the source completes, a later listener is still handed a closed stream. share(reset: false) is the 0.8.7 behaviour: the last cancel closes forever. A resubscribe needs a source that allows a second listen — Stream.fromIterable, Stream.multi, FxEvents.defer, a broadcast — a spent single-subscription StreamController still cannot be re-listened. Attach every listener before the first event if the source is one-shot, or keep one alive.

share() also does not remember: a listener that arrives after an event has passed has simply missed it. For a window of history, shareReplay is the next page. When latecomers need the current state — which is most UI — LiveValue is the answer, and LiveValue.from(source) / LiveValue.seededFrom(seed, source) build one directly from a stream. Those are hot: the subscription opens immediately, so values arriving before anyone listens still update value, and close() cancels the source. They are named constructors rather than an optional seed so that a nullable T can still be seeded with null. fxdart events layer, after Rx's share and shareValue.

Demo 1 · Why one listener is the default

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 · One run, two listeners

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

Try it yourself

Exercise: a LiveValue fed straight from a 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();
}

Related: shareReplay — multicast that remembers a buffer of history · LiveValue — the sharing that remembers: late subscribers get the current value first · tee — the pull-side answer to two readers over one pass, with no buffer · fork — two independent pull cursors over one source, at the cost of a buffer

share & LiveValue.from

Lecture

Demo 1 · Why one listener is the default

Demo 2 · One run, two listeners

Try it yourself

`share` & `LiveValue.from`