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 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;
     StreamSubscription subscription;
     controller = new StreamController(
       onListen: () {
         var add = controller.add;
         var addError = controller.addError;
         subscription = this.listen(
             (T event) {
@@ skipping 27 matching lines @@
   /**
    * 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 not a broadcast stream.
    */
   Stream asyncExpand(Stream convert(T event)) {
     StreamController controller;
     StreamSubscription subscription;
     controller = new StreamController(
       onListen: () {
         subscription = this.listen(
             (T event) {
               Stream newStream;
               try {
@@ skipping 36 matching lines @@
    * 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(); }
 }