Add setters for callbacks on StreamController.

sdk/lib/async/broadcast_stream_controller.dart

Index: sdk/lib/async/broadcast_stream_controller.dart
diff --git a/sdk/lib/async/broadcast_stream_controller.dart b/sdk/lib/async/broadcast_stream_controller.dart
index 93c0833317c26f2393dd14fb46aea127ca1edf4b..e5a1efc777d0e212c2ea3ae37b2622d024f5a071 100644
--- a/sdk/lib/async/broadcast_stream_controller.dart
+++ b/sdk/lib/async/broadcast_stream_controller.dart
@@ -67,7 +67,6 @@ class _BroadcastSubscription<T> extends _ControllerSubscription<T>
   // _onCancel is inherited.
 }
 
-
 abstract class _BroadcastStreamController<T>
     implements StreamController<T>,
                _StreamControllerLifecycle<T>,
@@ -80,8 +79,8 @@ abstract class _BroadcastStreamController<T>
   static const int _STATE_CLOSED = 4;
   static const int _STATE_ADDSTREAM = 8;
 
-  final _NotificationHandler _onListen;
-  final _NotificationHandler _onCancel;
+  _NotificationHandler _onListen;
+  _NotificationHandler _onCancel;
 
   // State of the controller.
   int _state;
@@ -113,6 +112,20 @@ abstract class _BroadcastStreamController<T>
     _next = _previous = this;
   }
 
+  void set onListen(void onListenHandler()) { _onListen = onListenHandler; }
+
+  void set onPause(void onPauseHandler()) {
+    throw new UnsupportedError(
+        "Broadcast stream controllers do not support pause callbacks");
+  }
+
+  void set onResume(void onResumeHandler())  {
+    throw new UnsupportedError(
+        "Broadcast stream controllers do not support pause callbacks");
+  }
+
+  void set onCancel(onCancelHandler()) { _onCancel = onCancelHandler; }
+
   // StreamController interface.
 
   Stream<T> get stream => new _BroadcastStream<T>(this);