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

sdk/lib/async/stream.dart

 // Copyright (c) 2013, the Dart project authors.  Please see the AUTHORS file
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
 part of dart.async;
 
 // -------------------------------------------------------------------
 // Core Stream types
 // -------------------------------------------------------------------
 
@@ skipping 31 matching lines @@
  * itself when the listener is unsubscribed, even if the stream wasn't done.
  *
  * Single-subscription streams are generally used for streaming parts of
  * contiguous data like file I/O.
  *
  * A broadcast stream allows any number of listeners, and it fires
  * its events when they are ready, whether there are listeners or not.
  *
  * 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.
  *
  * The default implementation of [isBroadcast] returns false.
  * A broadcast stream inheriting from [Stream] must override [isBroadcast]
  * to return [:true:].
  */
 abstract class Stream<T> {
   Stream();
 
@@ skipping 132 matching lines @@
   }
 
   /**
    * Reports whether this stream is a broadcast stream.
    */
   bool get isBroadcast => false;
 
   /**
    * 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.
    *
    * If [onListen] is provided, it is called with a subscription-like object
    * that represents the underlying subscription to this stream. It is
    * possible to pause, resume or cancel the subscription during the call
    * to [onListen]. It is not possible to change the event handlers, including
    * using [StreamSubscription.asFuture].
    *
    * If [onCancel] is provided, it is called in a similar way to [onListen]
    * when the returned stream stops having listener. If it later gets
    * a new listener, the [onListen] function is called again.
    *
    * Use the callbacks, for example, for pausing the underlying subscription
    * while having no subscribers to prevent losing events, or canceling the
    * subscription when there are no listeners.
    */
   Stream<T> asBroadcastStream({
       void onListen(StreamSubscription<T> subscription),
       void onCancel(StreamSubscription<T> subscription) }) {
-    if (isBroadcast) return this;
     return new _AsBroadcastStream<T>(this, onListen, onCancel);
   }
 
   /**
    * Adds a subscription to this stream.
    *
    * On each data event from this stream, the subscriber's [onData] handler
    * is called. If [onData] is null, nothing happens.
    *
    * On errors from this stream, the [onError] handler is given a
@@ skipping 14 matching lines @@
                                { Function onError,
                                  void onDone(),
                                  bool cancelOnError});
 
   /**
    * Creates a new stream from this stream that discards some data events.
    *
    * 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);
   }
 
   /**
    * 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);
   }
 
   /**
    * Creates a new stream with each data event of this stream asynchronously
    * mapped to a new event.
    *
    * 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 a broadcast stream if this stream is.
    */
   Stream asyncMap(convert(T event)) {
     StreamController controller;
     StreamSubscription subscription;
-    controller = new StreamController(
-      onListen: () {
-        var add = controller.add;
-        var addError = controller.addError;
-        subscription = this.listen(
-            (T event) {
-              var newValue;
-              try {
-                newValue = convert(event);
-              } catch (e, s) {
-                controller.addError(e, s);
-                return;
-              }
-              if (newValue is Future) {
-                subscription.pause();
-                newValue.then(add, onError: addError)
-                        .whenComplete(subscription.resume);
-              } else {
-                controller.add(newValue);
-              }
-            },
-            onError: addError,
-            onDone: controller.close
-        );
-      },
-      onPause: () { subscription.pause(); },
-      onResume: () { subscription.resume(); },
-      onCancel: () { subscription.cancel(); },
-      sync: true
-    );
+    void onListen () {
+      var add = controller.add;
+      var addError = controller.addError;
+      subscription = this.listen(
+          (T event) {
+            var newValue;
+            try {
+              newValue = convert(event);
+            } catch (e, s) {
+              controller.addError(e, s);
+              return;
+            }
+            if (newValue is Future) {
+              subscription.pause();
+              newValue.then(add, onError: addError)
+                      .whenComplete(subscription.resume);
+            } else {
+              controller.add(newValue);
+            }
+          },
+          onError: addError,
+          onDone: controller.close
+      );
+    }
+    if (this.isBroadcast) {
+      controller = new StreamController.broadcast(
+        onListen: onListen,
+        onCancel: () { subscription.cancel(); },
+        sync: true
+      );
+    } else {
+      controller = new StreamController(
+        onListen: onListen,
+        onPause: () { subscription.pause(); },
+        onResume: () { subscription.resume(); },
+        onCancel: () { subscription.cancel(); },
+        sync: true
+      );
+    }
     return controller.stream;
   }
 
   /**
    * Creates a new stream with the events of a stream per original event.
    *
    * This acts like [expand], except that [convert] returns a [Stream]
    * instead of an [Iterable].
    * The events of the returned stream becomes the events of the returned
    * stream, in the order they are produced.
    *
    * 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 broadcast stream if this stream is.
    */
   Stream asyncExpand(Stream convert(T event)) {
     StreamController controller;
     StreamSubscription subscription;
-    controller = new StreamController(
-      onListen: () {
-        subscription = this.listen(
-            (T event) {
-              Stream newStream;
-              try {
-                newStream = convert(event);
-              } catch (e, s) {
-                controller.addError(e, s);
-                return;
-              }
-              if (newStream != null) {
-                subscription.pause();
-                controller.addStream(newStream)
-                          .whenComplete(subscription.resume);
-              }
-            },
-            onError: controller.addError,
-            onDone: controller.close
-        );
-      },
-      onPause: () { subscription.pause(); },
-      onResume: () { subscription.resume(); },
-      onCancel: () { subscription.cancel(); },
-      sync: true
-    );
+    void onListen() {
+      subscription = this.listen(
+          (T event) {
+            Stream newStream;
+            try {
+              newStream = convert(event);
+            } catch (e, s) {
+              controller.addError(e, s);
+              return;
+            }
+            if (newStream != null) {
+              subscription.pause();
+              controller.addStream(newStream)
+                        .whenComplete(subscription.resume);
+            }
+          },
+          onError: controller.addError,
+          onDone: controller.close
+      );
+    }
+    if (this.isBroadcast) {
+      controller = new StreamController.broadcast(
+        onListen: onListen,
+        onCancel: () { subscription.cancel(); },
+        sync: true
+      );
+    } else {
+      controller = new StreamController(
+        onListen: onListen,
+        onPause: () { subscription.pause(); },
+        onResume: () { subscription.resume(); },
+        onCancel: () { subscription.cancel(); },
+        sync: true
+      );
+    }
     return controller.stream;
   }
 
   /**
    * Creates a wrapper Stream that intercepts some errors from this stream.
    *
    * If this stream sends an error that matches [test], then it is intercepted
    * by the [handle] function.
    *
    * The [onError] callback must be of type `void onError(error)` or
    * `void onError(error, StackTrace stackTrace)`. Depending on the function
    * type the the stream either invokes [onError] with or without a stack
    * trace. The stack trace argument might be `null` if the stream itself
    * received an error without stack trace.
    *
    * An asynchronous error [:e:] is matched by a test function if [:test(e):]
    * returns true. If [test] is omitted, every error is considered matching.
    *
    * If the error is intercepted, the [handle] function can decide what to do
    * with it. It can throw if it wants to raise a new (or the same) error,
    * or simply return to make the stream forget the error.
    *
    * If you need to transform an error into a data event, use the more generic
    * [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);
   }
 
   /**
    * Creates a new stream from this stream that converts each element
    * into zero or more events.
    *
    * Each incoming event is converted to an [Iterable] of new events,
    * 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);
   }
 
   /**
    * Binds this stream as the input of the provided [StreamConsumer].
    */
   Future pipe(StreamConsumer<T> streamConsumer) {
     return streamConsumer.addStream(this).then((_) => streamConsumer.close());
   }
 
   /**
    * 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);
   }
 
   /**
    * Reduces a sequence of values by repeatedly applying [combine].
    */
   Future<T> reduce(T combine(T previous, T element)) {
     _Future<T> result = new _Future<T>();
@@ skipping 299 matching lines @@
    * If this stream produces fewer than [count] values before it's done,
    * so will the returned stream.
    *
    * Stops listening to the stream after the first [n] elements have been
    * received.
    *
    * Internally the method cancels its subscription after these elements. This
    * 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);
   }
 
   /**
    * Forwards data events while [test] is successful.
    *
    * The returned stream provides the same events as this stream as long
    * as [test] returns [:true:] for the event data. The stream is done
    * when either this stream is done, or when this stream first provides
    * a value that [test] doesn't accept.
    *
    * Stops listening to the stream after the accepted elements.
    *
    * Internally the method cancels its subscription after these elements. This
    * 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);
   }
 
   /**
    * 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);
   }
 
   /**
    * Skip data events from this stream while they are matched by [test].
    *
    * Error and done events are provided by the returned stream unmodified.
    *
    * 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);
   }
 
   /**
    * Skips data events if they are equal to the previous data event.
    *
    * The returned stream provides the same events as this stream, except
    * that it never provides two consequtive data events that are equal.
    *
    * 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);
   }
 
   /**
    * Returns the first element of the stream.
    *
    * Stops listening to the stream after the first element has been received.
    *
@@ skipping 271 matching lines @@
    * The countdown is reset every time an event is forwarded from this stream,
    * or when the stream is paused and resumed.
    *
    * The [onTimeout] function is called with one argument: an
    * [EventSink] that allows putting events into the returned stream.
    * This `EventSink` is only valid during the call to `onTimeout`.
    *
    * 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)}) {
+    StreamController controller;
+    // The following variables are set on listen.
     StreamSubscription<T> subscription;
-    _StreamController controller;
-    // The following variables are set on listen.
     Timer timer;
     Zone zone;
     Function timeout;
 
     void onData(T event) {
       timer.cancel();
       controller.add(event);
       timer = zone.createTimer(timeLimit, timeout);
     }
     void onError(error, StackTrace stackTrace) {
       timer.cancel();
       controller.addError(error, stackTrace);
       timer = zone.createTimer(timeLimit, timeout);
     }
     void onDone() {
       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;
-            };
-          }
+    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);
-        },
-        () {
-          timer.cancel();
-          subscription.pause();
-        },
-        () {
-          subscription.resume();
-          timer = zone.createTimer(timeLimit, timeout);
-        },
-        () {
-          timer.cancel();
-          Future result = subscription.cancel();
-          subscription = null;
-          return result;
-        });
+      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;
   }
 }
 
 /**
  * A control object for the subscription on a [Stream].
  *
  * When you subscribe on a [Stream] using [Stream.listen],
  * a [StreamSubscription] object is returned. This object
  * is used to later unsubscribe again, or to temporarily pause
@@ skipping 324 matching lines @@
 class _ControllerEventSinkWrapper<T> implements EventSink<T> {
   EventSink _sink;
   _ControllerEventSinkWrapper(this._sink);
 
   void add(T data) { _sink.add(data); }
   void addError(error, [StackTrace stackTrace]) {
     _sink.addError(error, stackTrace);
   }
   void close() { _sink.close(); }
 }