Add getters for callbacks on StreamController.

sdk/lib/async/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;
 
 // -------------------------------------------------------------------
 // Controller for creating and adding events to a stream.
 // -------------------------------------------------------------------
 
 /**
+ * Type of a stream controller's `onListen`, `onPause` and `onResume` callbacks.
+ */
+typedef void ControllerCallback();
+
+/**
+ * Type of stream controller `onCancel` callbacks.
+ *
+ * The callback may return either `void` or a future.
+ */
+typedef ControllerCancelCallback();
+
+/**
  * A controller with the stream it controls.
  *
  * This controller allows sending data, error and done events on
  * its [stream].
  * This class can be used to create a simple stream that others
  * can listen on, and to push events to that stream.
  *
  * It's possible to check whether the stream is paused or not, and whether
  * it has subscribers or not, as well as getting a callback when either of
  * these change.
@@ skipping 104 matching lines @@
    */
   factory StreamController.broadcast({void onListen(),
                                       void onCancel(),
                                       bool sync: false}) {
     return sync
         ? new _SyncBroadcastStreamController<T>(onListen, onCancel)
         : new _AsyncBroadcastStreamController<T>(onListen, onCancel);
   }
 
   /**
-   * Sets the callback which is called when the stream is listened to.
+   * The callback which is called when the stream is listened to.
    *
-   * This overrides the previous callback, or clears it if the [onListenHandler]
-   * is `null`.
+   * May be set to `null`, in which case no callback will happen.
    */
+  ControllerCallback get onListen;
+
   void set onListen(void onListenHandler());
 
   /**
-   * Sets the callback which is called when the stream is paused.
+   * The callback which is called when the stream is paused.
    *
-   * This overrides the previous callback, or clears it if the [onPauseHandler]
-   * is `null`.
+   * May be set to `null`, in which case no callback will happen.
    *
    * Pause related callbacks are not supported on broadcast stream controllers.
    */
+  ControllerCallback get onPause;
+
   void set onPause(void onPauseHandler());
 
   /**
-   * Sets the callback which is called when the stream is resumed.
+   * The callback which is called when the stream is resumed.
    *
-   * This overrides the previous callback, or clears it if the [onResumeHandler]
-   * is `null`.
+   * May be set to `null`, in which case no callback will happen.
    *
    * Pause related callbacks are not supported on broadcast stream controllers.
    */
+  ControllerCallback get onResume;
+
   void set onResume(void onResumeHandler());
 
   /**
-   * Sets the callback which is called when the stream is canceled.
+   * The callback which is called when the stream is canceled.
    *
-   * This overrides the previous callback, or clears it if the [onCancelHandler]
-   * is `null`.
+   * May be set to `null`, in which case no callback will happen.
    */
+  ControllerCancelCallback get onCancel;
+
   void set onCancel(onCancelHandler());
 
   /**
    * Returns a view of this object that only exposes the [StreamSink] interface.
    */
   StreamSink<T> get sink;
 
   /**
    * Whether the stream controller is closed for adding more events.
    *
@@ skipping 227 matching lines @@
   /**
    * Future completed when the stream sends its last event.
    *
    * This is also the future returned by [close].
    */
   // TODO(lrn): Could this be stored in the varData field too, if it's not
   // accessed until the call to "close"? Then we need to special case if it's
   // accessed earlier, or if close is called before subscribing.
   _Future _doneFuture;
 
-  _NotificationHandler _onListen;
-  _NotificationHandler _onPause;
-  _NotificationHandler _onResume;
-  _NotificationHandler _onCancel;
+  ControllerCallback onListen;
+  ControllerCallback onPause;
+  ControllerCallback onResume;
+  ControllerCancelCallback onCancel;
 
-  _StreamController(void this._onListen(),
-                    void this._onPause(),
-                    void this._onResume(),
-                    this._onCancel());
-
-  void set onListen(void onListenHandler()) { _onListen = onListenHandler; }
-
-  void set onPause(void onPauseHandler()) { _onPause = onPauseHandler; }
-
-  void set onResume(void onResumeHandler()) { _onResume = onResumeHandler; }
-
-  void set onCancel(onCancelHandler()) { _onCancel = onCancelHandler; }
+  _StreamController(this.onListen,
+                    this.onPause,
+                    this.onResume,
+                    this.onCancel);
 
   // Return a new stream every time. The streams are equal, but not identical.
   Stream<T> get stream => new _ControllerStream<T>(this);
 
   /**
    * Returns a view of this object that only exposes the [StreamSink] interface.
    */
   StreamSink<T> get sink => new _StreamSinkWrapper<T>(this);
 
   /**
@@ skipping 203 matching lines @@
     _state |= _STATE_SUBSCRIBED;
     if (_isAddingStream) {
       _StreamControllerAddStreamState addState = _varData;
       addState.varData = subscription;
       addState.resume();
     } else {
       _varData = subscription;
     }
     subscription._setPendingEvents(pendingEvents);
     subscription._guardCallback(() {
-      _runGuarded(_onListen);
+      _runGuarded(onListen);
     });
 
     return subscription;
   }
 
   Future _recordCancel(StreamSubscription<T> subscription) {
     // When we cancel, we first cancel any stream being added,
-    // Then we call _onCancel, and finally the _doneFuture is completed.
-    // If either of addStream's cancel or _onCancel returns a future,
+    // Then we call `onCancel`, and finally the _doneFuture is completed.
+    // If either of addStream's cancel or `onCancel` returns a future,
     // we wait for it before continuing.
     // Any error during this process ends up in the returned future.
     // If more errors happen, we act as if it happens inside nested try/finallys
     // or whenComplete calls, and only the last error ends up in the
     // returned future.
     Future result;
     if (_isAddingStream) {
       _StreamControllerAddStreamState addState = _varData;
       result = addState.cancel();
     }
     _varData = null;
     _state =
         (_state & ~(_STATE_SUBSCRIBED | _STATE_ADDSTREAM)) | _STATE_CANCELED;
 
-    if (_onCancel != null) {
+    if (onCancel != null) {
       if (result == null) {
         // Only introduce a future if one is needed.
         // If _onCancel returns null, no future is needed.
         try {
-          result = _onCancel();
+          result = onCancel();
         } catch (e, s) {
           // Return the error in the returned future.
           // Complete it asynchronously, so there is time for a listener
           // to handle the error.
           result = new _Future().._asyncCompleteError(e, s);
         }
       } else {
         // Simpler case when we already know that we will return a future.
-        result = result.whenComplete(_onCancel);
+        result = result.whenComplete(onCancel);
       }
     }
 
     void complete() {
       if (_doneFuture != null && _doneFuture._mayComplete) {
         _doneFuture._asyncComplete(null);
       }
     }
 
     if (result != null) {
       result = result.whenComplete(complete);
     } else {
       complete();
     }
 
     return result;
   }
 
   void _recordPause(StreamSubscription<T> subscription) {
     if (_isAddingStream) {
       _StreamControllerAddStreamState addState = _varData;
       addState.pause();
     }
-    _runGuarded(_onPause);
+    _runGuarded(onPause);
   }
 
   void _recordResume(StreamSubscription<T> subscription) {
     if (_isAddingStream) {
       _StreamControllerAddStreamState addState = _varData;
       addState.resume();
     }
-    _runGuarded(_onResume);
+    _runGuarded(onResume);
   }
 }
 
 abstract class _SyncStreamControllerDispatch<T>
     implements _StreamController<T>, SynchronousStreamController<T> {
   int get _state;
   void set _state(int state);
 
   void _sendData(T data) {
     _subscription._add(data);
@@ skipping 170 matching lines @@
   _StreamControllerAddStreamState(_StreamController controller,
                                   this.varData,
                                   Stream source,
                                   bool cancelOnError)
       : super(controller, source, cancelOnError) {
     if (controller.isPaused) {
       addSubscription.pause();
     }
   }
 }