Stream transformations shouldn't preserve broadcast state

[nex3]

From the Stream documentation: "Stream transformations, such as where and skip, always return non-broadcast streams."

The following code snippet demonstrates that this isn't consistently true among transformation methods:

    void main() {
      var stream = new StreamController.broadcast().stream;
      testBroadcast("asyncExpand",
          stream.asyncExpand((e) => new Stream.fromIterable([e])));
      testBroadcast("asyncMap", stream.asyncMap((e) => e));
      testBroadcast("distinct", stream.distinct());
      testBroadcast("expand", stream.expand((e) => [e]));
      testBroadcast("handleError", stream.handleError((e) => throw e));
      testBroadcast("map", stream.map((e) => e));
      testBroadcast("skip", stream.skip(1));
      testBroadcast("skipWhile", stream.skipWhile(() => false));
      testBroadcast("takeWhile", stream.skipWhile(() => true));
      testBroadcast("timeout", stream.timeout(new Duration(seconds: 5)));
      testBroadcast("transform",
          stream.transform(new StreamTransformer.fromHandlers()));
      testBroadcast("where", stream.where((_) => true));
    }

    void testBroadcast(String name, Stream stream) {
      print("$name ${stream.isBroadcast ? 'is' : 'is not'} broadcast");
    }

This prints:

    asyncExpand is not broadcast
    asyncMap is not broadcast
    distinct is broadcast
    expand is broadcast
    handleError is broadcast
    map is broadcast
    skip is broadcast
    skipWhile is broadcast
    takeWhile is broadcast
    timeout is not broadcast
    transform is not broadcast
    where is broadcast

Most transformations, including "where" and "skip" which are explicitly called out as not preserving the broadcast state, do in fact preserve it.

[lrhn]

Either the documentation is wrong, or the implementation is.
I don't remember if we changed our mind here at some point. Florian?

---

cc @floitschG.

[lrhn]

It seems it is rather random right now.
Those transformations that simply wrap the subscription (map, etc.) preserve the type of the original stream.
Those that create a new stream (asyncMap/asyncExpand/timeout) generally create a single-subscription stream.

We should probably make async*/timeout inherit the stream type too, and change the documentation (that is the least breaking way to fix this).

The "transform" method depends on the transformation, which is reasonable. The StreamTransformer.fromHandlers could be changed to inherit its input stream's behavior.

---

Added Accepted label.

[nex3]

If the solution to this is to make all transformations inherit the original stream's type, please add a separate [asSingleSubscription] method so that there's a clean way to get a single-subscription stream from a broadcast stream. Please also make this method enqueue events as per issue #18588.

[floitschG]

A broadcast stream, by convention, has events that could potentially be dropped. If you actually want to collect them you should listen to the stream (note that a html-stream does not register itself to the DOM unless something is listening).
An asSingleSubscriptionStream is not a solution: a stream should not just explode in memory if nobody listens to it. The fact that single-subscription controllers buffer is one of the most-abused features, and we shouldn't make this even easier.

[nex3]

Sometimes buffering is the desired behavior. Document the pitfalls if you need to, but don't make a valid usage pattern unduly difficult because it's occasionally incorrect.

[lrhn]

The stream transformations have been documented to preserve the source stream's broadcastness. Those few that didn't have been modified to do so.

The asBroadcastStream still listens on the source stream when it is itself listened to. It is not just a transformation of the subscription like the transformations.
It has been changed to work the same for both broadcast and non-broadcast streams, instead of returning the stream itself it it was already broadcast.

---

Added Fixed label.