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)?}) =>
    _stream.asBroadcastStream(onListen: onListen, onCancel: onCancel);