asBroadcastStream 方法
- void onListen(
- StreamSubscription<
T> subscription
- StreamSubscription<
- void onCancel(
- StreamSubscription<
T> subscription
- StreamSubscription<
override
返回一个多订阅流,该流产生与该流相同的事件。
返回的流将在其第一个订阅者添加时订阅此流,并保持订阅状态,直到此流结束或回调取消订阅。
如果提供了 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)?}) =>
_stream.asBroadcastStream(onListen: onListen, onCancel: onCancel);