Make Stream.where, etc., be documented as inheriting broadcast state.

sdk/lib/async/stream.dart

Index: sdk/lib/async/stream.dart
diff --git a/sdk/lib/async/stream.dart b/sdk/lib/async/stream.dart
index a3b266f4f3b37b990dd777fe8fa1a4233bcd9e49..d94e62cb33db96710b7d258b52f63cf3650e21f8 100644
--- a/sdk/lib/async/stream.dart
+++ b/sdk/lib/async/stream.dart
@@ -49,8 +49,11 @@ part of dart.async;
  *
  * Broadcast streams are used for independent events/observers.
  *
- * Stream transformations, such as [where] and [skip], always return
- * non-broadcast streams. If several listeners want to listen to the returned
+ * Stream transformations, such as [where] and [skip],
+ * return the same type of stream as the one the method was called on,
+ * unless otherwise noted.
+ *
+ * If several listeners want to listen to the returned
  * stream, use [asBroadcastStream] to create a broadcast stream on top of the
  * non-broadcast stream.
  *
@@ -203,10 +206,7 @@ abstract class Stream<T> {
   /**
    * Returns a multi-subscription stream that produces the same events as this.
    *
-   * If this stream is already a broadcast stream, it is returned unmodified.
-   *
-   * If this stream is single-subscription, return a new stream that allows
-   * multiple subscribers. It will subscribe to this stream when its first
+   * The returned stream will subscribe to this stream when its first
    * subscriber is added, and will stay subscribed until this stream ends,
    * or a callback cancels the subscription.
    *
@@ -227,7 +227,6 @@ abstract class Stream<T> {
   Stream<T> asBroadcastStream({
       void onListen(StreamSubscription<T> subscription),
       void onCancel(StreamSubscription<T> subscription) }) {
-    if (isBroadcast) return this;
     return new _AsBroadcastStream<T>(this, onListen, onCancel);
   }
 
@@ -262,7 +261,7 @@ abstract class Stream<T> {
    * The new stream sends the same error and done events as this stream,
    * but it only sends the data events that satisfy the [test].
    *
-   * The returned stream is not a broadcast stream, even if this stream is.
+   * The returned stream is a broadcast stream if this stream is.
    */
   Stream<T> where(bool test(T event)) {
     return new _WhereStream<T>(this, test);
@@ -272,7 +271,7 @@ abstract class Stream<T> {
    * Creates a new stream that converts each element of this stream
    * to a new value using the [convert] function.
    *
-   * The returned stream is not a broadcast stream, even if this stream is.
+   * The returned stream is a broadcast stream if this stream is.
    */
   Stream map(convert(T event)) {
     return new _MapStream<T, dynamic>(this, convert);
@@ -285,6 +284,8 @@ abstract class Stream<T> {
    * This acts like [map], except that [convert] may return a [Future],
    * and in that case, the stream waits for that future to complete before
    * continuing with its result.
+   *
+   * The returned stream is not a broadcast stream.

[nweiz, 2014/05/20 21:15:54]

Clarify that despite this being a single-subscription stream, it will not buffer
events. Same for [asyncExpand].

[Lasse Reichstein Nielsen, 2014/05/21 06:16:53]

Technically, streams never buffer anything. Controllers and subscriptions may,
but streams don't do anything.

But yes, it's probably worth mentioning that this transformation doesn't listen
on the source until you listen to the result, and if the source is a broadcast
stream, events prior to that listen are not included.

    */
   Stream asyncMap(convert(T event)) {
     StreamController controller;
@@ -332,6 +333,8 @@ abstract class Stream<T> {
    *
    * If [convert] returns `null`, no value is put on the output stream,
    * just as if it returned an empty stream.
+   *
+   * The returned stream is a not a broadcast stream.
    */
   Stream asyncExpand(Stream convert(T event)) {
     StreamController controller;
@@ -388,7 +391,7 @@ abstract class Stream<T> {
    * [Stream.transform] to handle the event by writing a data event to
    * the output sink
    *
-   * The returned stream is not a broadcast stream, even if this stream is.
+   * The returned stream is a broadcast stream if this stream is.
    */
   Stream<T> handleError(Function onError, { bool test(error) }) {
     return new _HandleErrorStream<T>(this, onError, test);
@@ -402,7 +405,7 @@ abstract class Stream<T> {
    * and each of these new events are then sent by the returned stream
    * in order.
    *
-   * The returned stream is not a broadcast stream, even if this stream is.
+   * The returned stream is a broadcast stream if this stream is.
    */
   Stream expand(Iterable convert(T value)) {
     return new _ExpandStream<T, dynamic>(this, convert);
@@ -419,6 +422,9 @@ abstract class Stream<T> {
    * Chains this stream as the input of the provided [StreamTransformer].
    *
    * Returns the result of [:streamTransformer.bind:] itself.
+   *
+   * The `streamTransformer` can decide whether it wants to return a
+   * broadcast stream or not.
    */
   Stream transform(StreamTransformer<T, dynamic> streamTransformer) {
     return streamTransformer.bind(this);
@@ -738,7 +744,7 @@ abstract class Stream<T> {
    * means that single-subscription (non-broadcast) streams are closed and
    * cannot be reused after a call to this method.
    *
-   * The returned stream is not a broadcast stream, even if this stream is.
+   * The returned stream is a broadcast stream if this stream is.
    */
   Stream<T> take(int count) {
     return new _TakeStream(this, count);
@@ -758,7 +764,7 @@ abstract class Stream<T> {
    * means that single-subscription (non-broadcast) streams are closed and
    * cannot be reused after a call to this method.
    *
-   * The returned stream is not a broadcast stream, even if this stream is.
+   * The returned stream is a broadcast stream if this stream is.
    */
   Stream<T> takeWhile(bool test(T element)) {
     return new _TakeWhileStream(this, test);
@@ -767,7 +773,7 @@ abstract class Stream<T> {
   /**
    * Skips the first [count] data events from this stream.
    *
-   * The returned stream is not a broadcast stream, even if this stream is.
+   * The returned stream is a broadcast stream if this stream is.
    */
   Stream<T> skip(int count) {
     return new _SkipStream(this, count);
@@ -781,7 +787,7 @@ abstract class Stream<T> {
    * Starting with the first data event where [test] returns false for the
    * event data, the returned stream will have the same events as this stream.
    *
-   * The returned stream is not a broadcast stream, even if this stream is.
+   * The returned stream is a broadcast stream if this stream is.
    */
   Stream<T> skipWhile(bool test(T element)) {
     return new _SkipWhileStream(this, test);
@@ -796,7 +802,7 @@ abstract class Stream<T> {
    * Equality is determined by the provided [equals] method. If that is
    * omitted, the '==' operator on the last provided data element is used.
    *
-   * The returned stream is not a broadcast stream, even if this stream is.
+   * The returned stream is a broadcast stream if this stream is.
    */
   Stream<T> distinct([bool equals(T previous, T next)]) {
     return new _DistinctStream(this, equals);
@@ -1088,12 +1094,12 @@ abstract class Stream<T> {
    * If `onTimeout` is omitted, a timeout will just put a [TimeoutException]
    * into the error channel of the returned stream.
    *
-   * The returned stream is not a broadcast stream, even if this stream is.
+   * The returned stream is a broadcast stream if this stream is.
    */
   Stream timeout(Duration timeLimit, {void onTimeout(EventSink sink)}) {
-    StreamSubscription<T> subscription;
-    _StreamController controller;
+    StreamController controller;
     // The following variables are set on listen.
+    StreamSubscription<T> subscription;
     Timer timer;
     Zone zone;
     Function timeout;
@@ -1112,46 +1118,51 @@ abstract class Stream<T> {
       timer.cancel();
       controller.close();
     }
-    controller = new _SyncStreamController(
-        () {
-          // This is the onListen callback for of controller.
-          // It runs in the same zone that the subscription was created in.
-          // Use that zone for creating timers and running the onTimeout
-          // callback.
-          zone = Zone.current;
-          if (onTimeout == null) {
-            timeout = () {
-              controller.addError(new TimeoutException("No stream event",
-                                                       timeLimit));
-            };
-          } else {
-            onTimeout = zone.registerUnaryCallback(onTimeout);
-            _ControllerEventSinkWrapper wrapper =
-                new _ControllerEventSinkWrapper(null);
-            timeout = () {
-              wrapper._sink = controller;  // Only valid during call.
-              zone.runUnaryGuarded(onTimeout, wrapper);
-              wrapper._sink = null;
-            };
-          }
-
-          subscription = this.listen(onData, onError: onError, onDone: onDone);
-          timer = zone.createTimer(timeLimit, timeout);
-        },
-        () {
-          timer.cancel();
-          subscription.pause();
-        },
-        () {
-          subscription.resume();
-          timer = zone.createTimer(timeLimit, timeout);
-        },
-        () {
-          timer.cancel();
-          Future result = subscription.cancel();
-          subscription = null;
-          return result;
-        });
+    void onListen() {
+      // This is the onListen callback for of controller.
+      // It runs in the same zone that the subscription was created in.
+      // Use that zone for creating timers and running the onTimeout
+      // callback.
+      zone = Zone.current;
+      if (onTimeout == null) {
+        timeout = () {
+          controller.addError(new TimeoutException("No stream event",
+                                                   timeLimit));
+        };
+      } else {
+        onTimeout = zone.registerUnaryCallback(onTimeout);
+        _ControllerEventSinkWrapper wrapper =
+            new _ControllerEventSinkWrapper(null);
+        timeout = () {
+          wrapper._sink = controller;  // Only valid during call.
+          zone.runUnaryGuarded(onTimeout, wrapper);
+          wrapper._sink = null;
+        };
+      }
+
+      subscription = this.listen(onData, onError: onError, onDone: onDone);
+      timer = zone.createTimer(timeLimit, timeout);
+    }
+    Future onCancel() {
+      timer.cancel();
+      Future result = subscription.cancel();
+      subscription = null;
+      return result;
+    }
+    controller = isBroadcast
+        ? new _SyncBroadcastStreamController(onListen, onCancel)
+        : new _SyncStreamController(
+              onListen,
+              () {
+                // Don't null the timer, onCancel may call cancel again.
+                timer.cancel();
+                subscription.pause();
+              },
+              () {
+                subscription.resume();
+                timer = zone.createTimer(timeLimit, timeout);
+              },
+              onCancel);
     return controller.stream;
   }
 }