Let StreamSubscription.cancel return a Future.

sdk/lib/async/stream_controller.dart

Index: sdk/lib/async/stream_controller.dart
diff --git a/sdk/lib/async/stream_controller.dart b/sdk/lib/async/stream_controller.dart
index 42ff1eb372d04d521d40fe8bc14456e6437c1476..c9ececca5da3741ca054143b233d593a09dbf5f7 100644
--- a/sdk/lib/async/stream_controller.dart
+++ b/sdk/lib/async/stream_controller.dart
@@ -74,7 +74,7 @@ abstract class StreamController<T> implements StreamSink<T> {
   factory StreamController({void onListen(),
                             void onPause(),
                             void onResume(),
-                            void onCancel(),
+                            onCancel(),

[floitsch, 2013/10/12 18:53:57]

Needs comment.

[Lasse Reichstein Nielsen, 2013/10/14 11:32:33]

Can the return type be Future? 
It's always ok to return null, but it may not be type-correct to use a void
function :(

[Anders Johnsen, 2013/10/16 11:52:21]

Done.

[Anders Johnsen, 2013/10/16 11:52:21]

Done.

                             bool sync: false}) {
     if (onListen == null && onPause == null &&
         onResume == null && onCancel == null) {
@@ -175,7 +175,7 @@ abstract class _StreamControllerLifecycle<T> {
                                    bool cancelOnError);
   void _recordPause(StreamSubscription<T> subscription) {}
   void _recordResume(StreamSubscription<T> subscription) {}
-  void _recordCancel(StreamSubscription<T> subscription) {}
+  Future _recordCancel(StreamSubscription<T> subscription) => null;
 }
 
 /**
@@ -469,7 +469,7 @@ abstract class _StreamController<T> implements StreamController<T>,
     return subscription;
   }
 
-  void _recordCancel(StreamSubscription<T> subscription) {
+  Future _recordCancel(StreamSubscription<T> subscription) {
     if (_isAddingStream) {
       _StreamControllerAddStreamState addState = _varData;
       addState.cancel();
@@ -477,10 +477,11 @@ abstract class _StreamController<T> implements StreamController<T>,
     _varData = null;
     _state =
         (_state & ~(_STATE_SUBSCRIBED | _STATE_ADDSTREAM)) | _STATE_CANCELED;
-    _runGuarded(_onCancel);
+    var future = _runGuarded(_onCancel);

[floitsch, 2013/10/12 18:53:57]

Use type.

[Anders Johnsen, 2013/10/16 11:52:21]

Done.

     if (_doneFuture != null && _doneFuture._mayComplete) {
       _doneFuture._asyncComplete(null);
     }
+    return future;
   }
 
   void _recordPause(StreamSubscription<T> subscription) {
@@ -543,7 +544,7 @@ class _AsyncStreamController<T> extends _StreamController<T>
   _AsyncStreamController(void this._onListen(),
                          void this._onPause(),
                          void this._onResume(),
-                         void this._onCancel());
+                         this._onCancel());
 }
 
 class _SyncStreamController<T> extends _StreamController<T>
@@ -556,7 +557,7 @@ class _SyncStreamController<T> extends _StreamController<T>
   _SyncStreamController(void this._onListen(),
                         void this._onPause(),
                         void this._onResume(),
-                        void this._onCancel());
+                        this._onCancel());
 }
 
 abstract class _NoCallbacks {
@@ -574,10 +575,10 @@ typedef _NoCallbackSyncStreamController/*<T>*/ = _StreamController/*<T>*/
 
 typedef void _NotificationHandler();
 
-void _runGuarded(_NotificationHandler notificationHandler) {
-  if (notificationHandler == null) return;
+_runGuarded(_NotificationHandler notificationHandler) {

[floitsch, 2013/10/12 18:53:57]

"Future" as return type ?

[Anders Johnsen, 2013/10/16 11:52:21]

Done.

+  if (notificationHandler == null) return null;
   try {

[Lasse Reichstein Nielsen, 2013/10/14 11:32:33]

Is this not just
 Zone.current.runGuarded(notificationHandler)
?

[Anders Johnsen, 2013/10/16 11:52:21]

Not sure, Florian?

[floitsch, 2013/10/16 14:43:44]

Zone.current.runGuarded(notificationHandler) sounds right.

-    notificationHandler();
+    return notificationHandler();
   } catch (e, s) {
     Zone.current.handleUncaughtError(_asyncError(e, s));
   }
@@ -619,8 +620,8 @@ class _ControllerSubscription<T> extends _BufferingStreamSubscription<T> {
                           bool cancelOnError)
       : super(onData, onError, onDone, cancelOnError);
 
-  void _onCancel() {
-    _controller._recordCancel(this);
+  Future _onCancel() {
+    return _controller._recordCancel(this);
   }
 
   void _onPause() {
@@ -648,7 +649,7 @@ class _StreamSinkWrapper<T> implements StreamSink<T> {
  * Object containing the state used to handle [StreamController.addStream].
  */
 class _AddStreamState<T> {
-  // [_FutureImpl] returned by call to addStream.
+  // [_Future] returned by call to addStream.
   _Future addStreamFuture;
 
   // Subscription on stream argument to addStream.