`share` & `LiveValue.from`

체인 한 번의 실행, 여러 리스너 — 그리고 늦게 온 사람을 위해 최신 값을 기억하는 버전.

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

강의

이 섹션의 모든 연산자는 자기만의 StreamController를 만들므로, 돌려주는 체인은 단일 구독입니다: 두 번 리슨하면 두 번째 리스너는 StateError를 받습니다. 그 기본값은 의도적입니다 — 체인을 차갑게 유지해 누군가 소비하기 전까지 아무것도 실행되지 않게 하고, 리스너별 상태를 정직하게 지킵니다. 하지만 그것은 위젯 둘이 같은 debounce·throttle·switchMap 피드를 두 번 만들지 않고는 함께 볼 수 없다는 뜻이기도 합니다.

share()가 그것을 해결합니다. 첫 리스너에서 연결하고 그때부터 모든 리스너에게 방송하므로, 몇 명이 보고 있든 상류의 일은 한 번만 일어납니다. debounce 타이머, 소켓, 비싼 map — 구독자마다 하나가 아니라 전부 하나씩입니다.

share({reset: true}) — 기본값 — 은 이제 Rx의 ref-count 리셋과 같습니다. 마지막 리스너가 소스가 완료되기 전에 떠나면, 상류 구독은 취소되고 다음 리스너는 새 구독을 시작합니다. 소스가 완료된 뒤에는, 나중에 오는 리스너는 여전히 닫힌 스트림을 받습니다. share(reset: false)는 0.8.7의 동작입니다: 마지막 취소가 영영 닫습니다. 재구독에는 두 번째 리슨을 허용하는 소스가 필요합니다 — Stream.fromIterable, Stream.multi, FxEvents.defer, 브로드캐스트 — 이미 소진된 단일 구독 StreamController는 여전히 다시 들을 수 없습니다. 소스가 한 번뿐이면 첫 이벤트 전에 모든 리스너를 붙이거나, 하나를 살려 두세요.

share()는 기억하지도 않습니다: 이벤트가 지나간 뒤에 도착한 리스너는 그냥 놓친 것입니다. 이력의 창이 필요하면 shareReplay가 다음 쪽입니다. 늦게 온 쪽이 현재 상태를 알아야 할 때 — 대부분의 UI가 그렇습니다 — LiveValue가 답이고, LiveValue.from(source)과 LiveValue.seededFrom(seed, source)이 스트림에서 바로 하나를 만듭니다. 이들은 뜨겁습니다: 구독이 즉시 열리므로 아무도 리슨하기 전에 도착한 값도 value를 갱신하고, close()는 소스를 취소합니다. 선택적 시드가 아니라 이름 붙은 생성자인 것은 널 가능한 T도 null로 시드할 수 있게 하기 위해서입니다. fxdart 이벤트 레이어, Rx의 share와 shareValue를 따랐습니다.

데모 1 · 왜 리스너 하나가 기본값인가

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

데모 2 · 한 번의 실행, 두 리스너

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

직접 해 보기

연습: 스트림에서 바로 만든 LiveValue.

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

관련 항목: shareReplay — 이력 버퍼를 기억하는 멀티캐스트 · LiveValue — 기억하는 공유: 늦은 구독자가 현재 값을 먼저 받음 · tee — 한 번의 순회에 독자 둘, 버퍼 없이 — 풀 쪽의 답 · fork — 한 소스에 독립적인 풀 커서 둘, 버퍼를 대가로

share & LiveValue.from

강의

데모 1 · 왜 리스너 하나가 기본값인가

데모 2 · 한 번의 실행, 두 리스너

직접 해 보기

`share` & `LiveValue.from`