Make asBroadcastStream take two callbacks.

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 131 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
-   * subscriber is added, and unsubscribe again when the last subscription is
-   * canceled.
+   * subscriber is added, and will stay subscribed until this stream ends,
+   * or a callback cancels the subscription.
    *
-   * If this stream is already a broadcast stream, it is returned unmodified.
+   * 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() {
+  Stream<T> asBroadcastStream({
+      void onListen(StreamSubscription<T> subscription),
+      void onCancel(StreamSubscription<T> subscription) }) {
     if (isBroadcast) return this;
-    return new _AsBroadcastStream<T>(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
    * object describing the error.
@@ skipping 730 matching lines @@
 
 
 /** [Stream] wrapper that only exposes the [Stream] interface. */
 class StreamView<T> extends Stream<T> {
   Stream<T> _stream;
 
   StreamView(this._stream);
 
   bool get isBroadcast => _stream.isBroadcast;
 
-  Stream<T> asBroadcastStream() => _stream.asBroadcastStream();
+  Stream<T> asBroadcastStream({void onListen(StreamSubscription subscription),
+                               void onCancel(StreamSubscription subscription)})
+      => _stream.asBroadcastStream(onListen: onListen, onCancel: onCancel);
 
   StreamSubscription<T> listen(void onData(T value),
                                { void onError(error),
                                  void onDone(),
                                  bool cancelOnError }) {
     return _stream.listen(onData, onError: onError, onDone: onDone,
                           cancelOnError: cancelOnError);
   }
 }
 
@@ skipping 318 matching lines @@
    *
    * If a [moveNext] call has been made, it will complete with `false` as value,
    * as will all further calls to [moveNext].
    *
    * If you need to stop listening for values before the stream iterator is
    * automatically closed, you must call [cancel] to ensure that the stream
    * is properly closed.
    */
   void cancel();
 }