Add getters 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 61 matching lines @@
                _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;
 
-  _NotificationHandler _onListen;
-  _NotificationHandler _onCancel;
+  ControllerCallback onListen;
+  ControllerCancelCallback 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;
 
   /**
    * Future returned by [close] and [done].
    *
    * The future is completed whenever the done event has been sent to all
    * relevant listeners.
    * The relevant listeners are the ones that were listening when [close] was
    * called. When all of these have been canceled (sending the done event makes
    * them cancel, but they can also be canceled before sending the event),
    * this future completes.
    *
    * Any attempt to listen after calling [close] will throw, so there won't
    * be any further listeners.
    */
   _Future _doneFuture;
 
-  _BroadcastStreamController(this._onListen, this._onCancel)
+  _BroadcastStreamController(this.onListen, this.onCancel)
       : _state = _STATE_INITIAL {
     _next = _previous = this;
   }
 
-  void set onListen(void onListenHandler()) { _onListen = onListenHandler; }
+  ControllerCallback get onPause {
+    throw new UnsupportedError(
+        "Broadcast stream controllers do not support pause callbacks");
+  }
 
   void set onPause(void onPauseHandler()) {
     throw new UnsupportedError(
         "Broadcast stream controllers do not support pause callbacks");
   }
 
+  ControllerCallback get onResume {
+    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 63 matching lines @@
     if (isClosed) {
       if (onDone == null) onDone = _nullDoneHandler;
       return new _DoneStreamSubscription<T>(onDone);
     }
     StreamSubscription subscription =
         new _BroadcastSubscription<T>(this, onData, onError, onDone,
                                       cancelOnError);
     _addListener(subscription);
     if (identical(_next, _previous)) {
       // Only one listener, so it must be the first listener.
-      _runGuarded(_onListen);
+      _runGuarded(onListen);
     }
     return subscription;
   }
 
   Future _recordCancel(_BroadcastSubscription<T> subscription) {
     // If already removed by the stream, don't remove it again.
     if (identical(subscription._next, subscription)) return null;
     assert(!identical(subscription._next, subscription));
     if (subscription._isFiring) {
       subscription._setRemoveAfterFiring();
@@ skipping 80 matching lines @@
   void _forEachListener(
       void action(_BufferingStreamSubscription<T> subscription)) {
     if (_isFiring) {
       throw new StateError(
           "Cannot fire new event. Controller is already firing an event");
     }
     if (_isEmpty) return;
 
     // Get event id of this event.
     int id = (_state & _STATE_EVENT_ID);
-    // Start firing (set the _STATE_FIRING bit). We don't do [_onCancel]
+    // Start firing (set the _STATE_FIRING bit). We don't do [onCancel]
     // callbacks while firing, and we prevent reentrancy of this function.
     //
     // Set [_state]'s event id to the next event's id.
     // Any listeners added while firing this event will expect the next event,
     // not this one, and won't get notified.
     _state ^= _STATE_EVENT_ID | _STATE_FIRING;
     _BroadcastSubscriptionLink link = _next;
     while (!identical(link, this)) {
       _BroadcastSubscription<T> subscription = link;
       if (subscription._expectsEvent(id)) {
@@ skipping 15 matching lines @@
       _callOnCancel();
     }
   }
 
   void _callOnCancel() {
     assert(_isEmpty);
     if (isClosed && _doneFuture._mayComplete) {
       // When closed, _doneFuture is not null.
       _doneFuture._asyncComplete(null);
     }
-    _runGuarded(_onCancel);
+    _runGuarded(onCancel);
   }
 }
 
 class _SyncBroadcastStreamController<T> extends _BroadcastStreamController<T>
                                      implements SynchronousStreamController<T> {
   _SyncBroadcastStreamController(void onListen(), void onCancel())
       : super(onListen, onCancel);
 
   // EventDispatch interface.
 
@@ skipping 167 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();
 }