dart:async

метод asBroadcastStream

Возвращает поток с несколькими подписками, который генерирует те же события, что и этот поток.

Возвращённый поток подпишется на этот поток при добавлении первого подписчика и останется подписанным, пока этот поток не завершится или пока обратный вызов не отменит подписку.

Если предоставлен onListen, он вызывается с объектом, подобным подписке, который представляет базовую подписку на этот поток. Во время вызова onListen можно приостановить, возобновить или отменить подписку. Изменить обработчики событий нельзя, в том числе с помощью StreamSubscription.asFuture.

Если предоставлен onCancel, он вызывается аналогично onListen, когда у возвращённого потока не остаётся слушателей. Если позднее у него появится новый слушатель, функция onListen будет вызвана снова.

Используйте обратные вызовы, например, чтобы приостанавливать базовую подписку, когда нет подписчиков, и не терять события, или отменять подписку, когда слушателей нет.

Отмена предназначена для случаев, когда текущих подписчиков нет. Если подписка, переданная в onListen или onCancel, отменена, текущие подписки на возвращённый широковещательный поток больше никогда не будут получать события, даже событие done.

Пример:

final stream =
    Stream<int>.periodic(const Duration(seconds: 1), (count) => count)
        .take(10);

final broadcastStream = stream.asBroadcastStream(
  onCancel: (controller) {
    print('Stream paused');
    controller.pause();
  },
  onListen: (controller) async {
    if (controller.isPaused) {
      print('Stream resumed');
      controller.resume();
    }
  },
);

final oddNumberStream = broadcastStream.where((event) => event.isOdd);
final oddNumberListener = oddNumberStream.listen(
      (event) {
    print('Odd: $event');
  },
  onDone: () => print('Done'),
);

final evenNumberStream = broadcastStream.where((event) => event.isEven);
final evenNumberListener = evenNumberStream.listen((event) {
  print('Even: $event');
}, onDone: () => print('Done'));

await Future.delayed(const Duration(milliseconds: 3500)); // 3.5 second
// Outputs:
// Even: 0
// Odd: 1
// Even: 2
oddNumberListener.cancel(); // Nothing printed.
evenNumberListener.cancel(); // "Stream paused"
await Future.delayed(const Duration(seconds: 2));
print(await broadcastStream.first); // "Stream resumed"
// Outputs:
// 3

Реализация

Stream<T> asBroadcastStream(
    {void onListen(StreamSubscription<T> subscription)?,
    void onCancel(StreamSubscription<T> subscription)?}) {
  return new _AsBroadcastStream<T>(this, onListen, onCancel);
}