Add setters for callbacks on StreamController.

sdk/lib/async/broadcast_stream_controller.dart

 // Copyright (c) 2012, 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;
 
 class _BroadcastStream<T> extends _ControllerStream<T> {
   _BroadcastStream(_StreamControllerLifecycle controller) : super(controller);
 
   bool get isBroadcast => true;
@@ skipping 49 matching lines @@
   // so we don't bother calling it.
   void _onPause() { }
 
   // The controller._recordResume doesn't do anything for a broadcast
   // controller, so we don't bother calling it.
   void _onResume() { }
 
   // _onCancel is inherited.
 }
 
-
 abstract class _BroadcastStreamController<T>
     implements StreamController<T>,
                _StreamControllerLifecycle<T>,
                _BroadcastSubscriptionLink,
                _EventSink<T>,
                _EventDispatch<T> {
   static const int _STATE_INITIAL = 0;
   static const int _STATE_EVENT_ID = 1;
   static const int _STATE_FIRING = 2;
   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;
 
   // Double-linked list of active listeners.
   _BroadcastSubscriptionLink _next;
   _BroadcastSubscriptionLink _previous;
 
   // Extra state used during an [addStream] call.
   _AddStreamState<T> _addStreamState;
@@ skipping 11 matching lines @@
    * Any attempt to listen after calling [close] will throw, so there won't
    * be any further listeners.
    */
   _Future _doneFuture;
 
   _BroadcastStreamController(this._onListen, this._onCancel)
       : _state = _STATE_INITIAL {
     _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);
 
   StreamSink<T> get sink => new _StreamSinkWrapper<T>(this);
 
   bool get isClosed => (_state & _STATE_CLOSED) != 0;
 
   /**
    * A broadcast controller is never paused.
@@ skipping 388 matching lines @@
     _pauseCount++;
   }
   void resume() { _resume(null); }
   void _resume(_) {
     if (_pauseCount > 0) _pauseCount--;
   }
   Future cancel() { return new _Future.immediate(null); }
   bool get isPaused => _pauseCount > 0;
   Future asFuture([Object value]) => new _Future();
 }