dart:async

Метод asBroadcastStream

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

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

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

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

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

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

Пример:

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