Stream.asBroadcastStream - Easy to cause leaks, what is the rationale?

[matanlurey]

As detailed a bit in the original commit (1315656), Stream#asBroadcastStream has behavior in which:

1. Listening to the (Broadcast) stream to the first time opens a subscription to the underlying stream
2. Closing the last subscription to the (Broadcast) stream does not close the subscription

The only way to close the underlying subscription is to receive a callback in asBroadcastStream:

fooStream.asBroadcastStream(onCancel: (subHandle) {
  subHandle.cancel();
});

Internal app ran into a big memory leak because of this, so I'm just wondering what the rationale is for making it so easy to leak.

@floitschG @lrhn @cbracken

[lrhn]

tl;dr: Because that's how broadcast streams work, and you might be using them for something they are not intended for.

Broadcast streams are intended to be streams that exist independently of their listeners. That's where the name comes from - they broadcast events into the "ether" whether anybody is listening or not. Canceling the last listener on a broadcast stream doesn't actually mean anything. Broadcast streams are intended for cases like DOM event handlers, where each event can be handled individually, it doesn't matter if you miss some of them, so you can start (and stop) listening at any point.

Normally, the code that controls the stream will know how to handle the lack of listeners. Code using a broadcast StreamController can choose to stop the source of its events at any point, for example when there are no listeners. That's normal and recommended behavior, there is no reason to create events while nobody is listening. If the broadcast stream is, e.g., replaying timer ticks, it can stop the timer entirely while there are no listeners, and start it again when someone starts listening.

For a broadcast stream created by calling asBroadcastStream on a single subscription stream, the options are limited. The default behavior is to treat the source stream as a continuous stream of events and broadcast all of them, closing when the original stream stops. That is, it really broadcasts the original stream's events.

The onListen and onCancel callbacks that you can pass to asBroadcastStream give you a chance to change the default behavior by either pausing or cancelling the original stream when there are no listeners. Pausing/resuming and canceling are the only options, since that's all you can do on a stream subscription.

We have the problem that broadcast streams, and in particular ones created using asBroadcastStream, are often used just to make a stream accept more than one listener. That's dangerous, because that isn't the only difference between broadcast streams and non-broadcast ("single-subscription") streams, and sometimes the other differences makes the code not work as expected.

If you want two streams that has the same events as one original stream, then you can use https://www.dartdocs.org/documentation/async/1.11.0/async/StreamSplitter-class.html from package:async. That also has memory leak issues if the streams aren't consumed at the same pace - it has to keep the events in memory until every split stream has delivered them.

If you want to have different listeners handle the events over time, you can either keep using the same subscription and changing the callback, or use the https://www.dartdocs.org/documentation/async/1.11.0/async/StreamQueue-class.html class, also from package:async, to programmatically distribute the events to the code that needs them.

[anders-sandholm]

Closing with "works as intended". Feel free to re-open if I'm mistaken.

[matanlurey]

That was an excellent explanation, thank you. Sounds very reasonable.