Make asBroadcastStream take two callbacks.

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 bff0435af274aef86df2fb9a4f2bcb14d61df57c..e6f162cfd5a58d66fedc52d9657d5fe319c1ae71 100644
--- a/sdk/lib/async/stream.dart
+++ b/sdk/lib/async/stream.dart
@@ -149,16 +149,32 @@ 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
-   * 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);
   }
 
   /**
@@ -909,7 +925,9 @@ class StreamView<T> extends Stream<T> {
 
   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),