asBroadcastStream 方法

Stream<T> asBroadcastStream(
  1. {void onListen(
    1. StreamSubscription<T> subscription
    )?,
  2. void onCancel(
    1. StreamSubscription<T> subscription
    )?}
)
override

返回一个多订阅流,其产生的事件与这个流相同。

返回的流将在第一个订阅者添加时订阅此流,并且将保持订阅,直到此流结束或回调取消订阅。

如果提供了 onListen,则它会用表示对当前流的底层订阅的类似订阅的对象调用。在调用 onListen 期间,可以暂停、恢复或取消订阅。不能更改事件处理程序,包括使用 StreamSubscription.asFuture

如果提供了 onCancel,则在返回的流停止有监听器时以类似 onListen 的方式调用。如果它稍后获得新的监听器,则再次调用 onListen 函数。

使用回调,例如,在没有任何订阅者时暂停底层订阅以避免丢失事件,或者在没有监听器时取消订阅。

取消订阅的目的是用于没有当前订阅者的情况。如果传递给 onListenonCancel 的订阅被取消,则返回的广播流上的当前订阅将永远不会发出任何进一步的事件,甚至不包括完成事件。

示例

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