Change StreamController constructor.

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.
 // -------------------------------------------------------------------
 
@@ skipping 32 matching lines @@
  *
  * If listeners are added after the stream has completed (sent a "done" event),
  * the listeners will be sent a "done" event eventually, but they won't affect
  * the stream at all, and won't trigger callbacks. From the controller's point
  * of view, the stream is completely inert when has completed.
  */
 class StreamController<T> extends EventSink<T> {
   final _StreamImpl<T> stream;
 
   /**
-   * A controller with a broadcast [stream]..
+   * A controller with a broadcast [stream].
    *
-   * The [onPauseStateChange] function is called when the stream becomes
-   * paused or resumes after being paused. The current pause state can
-   * be read from [isPaused]. Ignored if [:null:].
+   * The [onPause] function is called when the stream becomes
+   * paused. [onResume] is called when the stream resumed.
    *
-   * The [onSubscriptionStateChange] function is called when the stream
-   * receives its first listener or loses its last. The current subscription
-   * state can be read from [hasListener]. Ignored if [:null:].
+   * The [onListen] callback is called when the stream
+   * receives its first listener. [onCancel] when the last listener cancels
+   * its subscription.
+   *
+   * If the stream is canceled before the controller needs new data the
+   * [onResume] call might not be executed.
    */
-  StreamController.broadcast({void onPauseStateChange(),
-                              void onSubscriptionStateChange()})
-      : stream = new _MultiControllerStream<T>(onSubscriptionStateChange,
-                                               onPauseStateChange);
+  StreamController.broadcast({void onListen(),
+                              void onPause(),
+                              void onResume(),
+                              void onCancel()})
+      : stream = new _MultiControllerStream<T>(
+            onListen, onPause, onResume, onCancel);
 
   /**
    * A controller with a [stream] that supports only one single subscriber.
    *
    * The controller will buffer all incoming events until the subscriber is
    * registered.
    *
-   * The [onPauseStateChange] function is called when the stream becomes
-   * paused or resumes after being paused. The current pause state can
-   * be read from [isPaused]. Ignored if [:null:].
+   * The [onPause] function is called when the stream becomes
+   * paused. [onResume] is called when the stream resumed.
    *
-   * The [onSubscriptionStateChange] function is called when the stream
-   * receives its first listener or loses its last. The current subscription
-   * state can be read from [hasListener]. Ignored if [:null:].
+   * The [onListen] callback is called when the stream
+   * receives its listener. [onCancel] when the listener cancels
+   * its subscription.
+   *
+   * If the stream is canceled before the controller needs new data the
+   * [onResume] call might not be executed.
    */
-  StreamController({void onPauseStateChange(),
-                    void onSubscriptionStateChange()})
-      : stream = new _SingleControllerStream<T>(onSubscriptionStateChange,
-                                                onPauseStateChange);
+  StreamController({void onListen(),
+                    void onPause(),
+                    void onResume(),
+                    void onCancel()})
+      : stream = new _SingleControllerStream<T>(
+            onListen, onPause, onResume, onCancel);
 
   /**
    * Returns a view of this object that only exposes the [EventSink] interface.
    */
   EventSink<T> get sink => new _EventSinkView<T>(this);
 
   /**
    * Whether the stream is closed for adding more events.
    *
    * If true, the "done" event might not have fired yet, but it has been
@@ skipping 39 matching lines @@
    *
    * The "done" message should be sent at most once by a stream, and it
    * should be the last message sent.
    */
   void close() { stream._close(); }
 }
 
 typedef void _NotificationHandler();
 
 class _MultiControllerStream<T> extends _MultiStreamImpl<T> {
-  _NotificationHandler _subscriptionHandler;
-  _NotificationHandler _pauseHandler;
+  _NotificationHandler _onListen;
+  _NotificationHandler _onPause;
+  _NotificationHandler _onResume;
+  _NotificationHandler _onCancel;
 
-  _MultiControllerStream(this._subscriptionHandler, this._pauseHandler);
-
-  void _onSubscriptionStateChange() {
-    if (_subscriptionHandler != null) {
-      try {
-        _subscriptionHandler();
-      } catch (e, s) {
-        new AsyncError(e, s).throwDelayed();
-      }
+  // TODO(floitsch): share this code with _SingleControllerStream.
+  void _runGuarded(_NotificationHandler notificationHandler) {
+    if (notificationHandler == null) return;
+    try {
+      notificationHandler();
+    } catch (e, s) {
+      new AsyncError(e, s).throwDelayed();
     }
   }
 
+  _MultiControllerStream(this._onListen,
+                         this._onPause,
+                         this._onResume,
+                         this._onCancel);
+
+  void _onSubscriptionStateChange() {
+    _runGuarded(_hasListener ? _onListen : _onCancel);
+  }
+
   void _onPauseStateChange() {
-    if (_pauseHandler != null) {
-      try {
-        _pauseHandler();
-      } catch (e, s) {
-        new AsyncError(e, s).throwDelayed();
-      }
-    }
+    _runGuarded(_isPaused ? _onPause : _onResume);
   }
 }
 
 class _SingleControllerStream<T> extends _SingleStreamImpl<T> {
-  _NotificationHandler _subscriptionHandler;
-  _NotificationHandler _pauseHandler;
+  _NotificationHandler _onListen;
+  _NotificationHandler _onPause;
+  _NotificationHandler _onResume;
+  _NotificationHandler _onCancel;
 
-  _SingleControllerStream(this._subscriptionHandler, this._pauseHandler);
-
-  void _onSubscriptionStateChange() {
-    if (_subscriptionHandler != null) {
-      try {
-        _subscriptionHandler();
-      } catch (e, s) {
-        new AsyncError(e, s).throwDelayed();
-      }
+  // TODO(floitsch): share this code with _MultiControllerStream.
+  _runGuarded(_NotificationHandler notificationHandler) {
+    if (notificationHandler == null) return;
+    try {
+      notificationHandler();
+    } catch (e, s) {
+      new AsyncError(e, s).throwDelayed();
     }
   }
 
+  _SingleControllerStream(this._onListen,
+                          this._onPause,
+                          this._onResume,
+                          this._onCancel);
+
+  void _onSubscriptionStateChange() {
+    _runGuarded(_hasListener ? _onListen : _onCancel);
+  }
+
   void _onPauseStateChange() {
-    if (_pauseHandler != null) {
-      try {
-        _pauseHandler();
-      } catch (e, s) {
-        new AsyncError(e, s).throwDelayed();
-      }
-    }
+    _runGuarded(_isPaused ? _onPause : _onResume);
   }
 }