Let StreamSubscription.cancel return a Future.

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 56 matching lines @@
    * The [onListen] callback is called when the stream
    * receives its listener and [onCancel] when the listener ends
    * its subscription.
    *
    * If the stream is canceled before the controller needs new data the
    * [onResume] call might not be executed.
    */
   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) {
       return sync
           ? new _NoCallbackSyncStreamController/*<T>*/()
           : new _NoCallbackAsyncStreamController/*<T>*/();
     }
     return sync
          ? new _SyncStreamController<T>(onListen, onPause, onResume, onCancel)
          : new _AsyncStreamController<T>(onListen, onPause, onResume, onCancel);
@@ skipping 80 matching lines @@
 }
 
 
 abstract class _StreamControllerLifecycle<T> {
   StreamSubscription<T> _subscribe(void onData(T data),
                                    void onError(Object error),
                                    void onDone(),
                                    bool cancelOnError);
   void _recordPause(StreamSubscription<T> subscription) {}
   void _recordResume(StreamSubscription<T> subscription) {}
-  void _recordCancel(StreamSubscription<T> subscription) {}
+  Future _recordCancel(StreamSubscription<T> subscription) => null;
 }
 
 /**
  * Default implementation of [StreamController].
  *
  * Controls a stream that only supports a single controller.
  */
 abstract class _StreamController<T> implements StreamController<T>,
                                                _StreamControllerLifecycle<T>,
                                                _EventSink<T>,
@@ skipping 273 matching lines @@
       _varData = subscription;
     }
     subscription._setPendingEvents(pendingEvents);
     subscription._guardCallback(() {
       _runGuarded(_onListen);
     });
 
     return subscription;
   }
 
-  void _recordCancel(StreamSubscription<T> subscription) {
+  Future _recordCancel(StreamSubscription<T> subscription) {
     if (_isAddingStream) {
       _StreamControllerAddStreamState addState = _varData;
       addState.cancel();
     }
     _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) {
     if (_isAddingStream) {
       _StreamControllerAddStreamState addState = _varData;
       addState.pause();
     }
     _runGuarded(_onPause);
   }
 
@@ skipping 42 matching lines @@
 class _AsyncStreamController<T> extends _StreamController<T>
                                    with _AsyncStreamControllerDispatch<T> {
   final _NotificationHandler _onListen;
   final _NotificationHandler _onPause;
   final _NotificationHandler _onResume;
   final _NotificationHandler _onCancel;
 
   _AsyncStreamController(void this._onListen(),
                          void this._onPause(),
                          void this._onResume(),
-                         void this._onCancel());
+                         this._onCancel());
 }
 
 class _SyncStreamController<T> extends _StreamController<T>
                                   with _SyncStreamControllerDispatch<T> {
   final _NotificationHandler _onListen;
   final _NotificationHandler _onPause;
   final _NotificationHandler _onResume;
   final _NotificationHandler _onCancel;
 
   _SyncStreamController(void this._onListen(),
                         void this._onPause(),
                         void this._onResume(),
-                        void this._onCancel());
+                        this._onCancel());
 }
 
 abstract class _NoCallbacks {
   _NotificationHandler get _onListen => null;
   _NotificationHandler get _onPause => null;
   _NotificationHandler get _onResume => null;
   _NotificationHandler get _onCancel => null;
 }
 
 typedef _NoCallbackAsyncStreamController/*<T>*/ = _StreamController/*<T>*/
        with _AsyncStreamControllerDispatch/*<T>*/, _NoCallbacks;
 
 typedef _NoCallbackSyncStreamController/*<T>*/ = _StreamController/*<T>*/
        with _SyncStreamControllerDispatch/*<T>*/, _NoCallbacks;
 
 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));
   }
 }
 
 class _ControllerStream<T> extends _StreamImpl<T> {
   _StreamControllerLifecycle<T> _controller;
 
   _ControllerStream(this._controller);
 
@@ skipping 21 matching lines @@
 class _ControllerSubscription<T> extends _BufferingStreamSubscription<T> {
   final _StreamControllerLifecycle<T> _controller;
 
   _ControllerSubscription(this._controller,
                           void onData(T data),
                           void onError(Object error),
                           void onDone(),
                           bool cancelOnError)
       : super(onData, onError, onDone, cancelOnError);
 
-  void _onCancel() {
-    _controller._recordCancel(this);
+  Future _onCancel() {
+    return _controller._recordCancel(this);
   }
 
   void _onPause() {
     _controller._recordPause(this);
   }
 
   void _onResume() {
     _controller._recordResume(this);
   }
 }
 
 
 /** A class that exposes only the [StreamSink] interface of an object. */
 class _StreamSinkWrapper<T> implements StreamSink<T> {
   final StreamSink _target;
   _StreamSinkWrapper(this._target);
   void add(T data) { _target.add(data); }
   void addError(Object error) { _target.addError(error); }
   Future close() => _target.close();
   Future addStream(Stream<T> source) => _target.addStream(source);
   Future get done => _target.done;
 }
 
 /**
  * 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.
   StreamSubscription addSubscription;
 
   _AddStreamState(_EventSink<T> controller, Stream source)
       : addStreamFuture = new _Future(),
         addSubscription = source.listen(controller._add,
                                         onError: controller._addError,
                                         onDone: controller._close,
@@ skipping 24 matching lines @@
   var varData;
 
   _StreamControllerAddStreamState(_StreamController controller,
                                   this.varData,
                                   Stream source) : super(controller, source) {
     if (controller.isPaused) {
       addSubscription.pause();
     }
   }
 }