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 48 matching lines @@
    * the event has returned.
    *
    * The controller will buffer all incoming events until the subscriber is
    * registered.
    *
    * The [onPause] function is called when the stream becomes
    * paused. [onResume] is called when the stream resumed.
    *
    * The [onListen] callback is called when the stream
    * receives its listener and [onCancel] when the listener ends
-   * its subscription.
+   * its subscription. If [onCancel] needs to perform an asynchronous operation,
+   * [onCancel] can return a [Future]. The [StreamSubscription] will then get

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

should return a future that completes when the cancel operation is done.

Remove the remainder of the paragraph.

[Anders Johnsen, 2013/10/21 08:01:46]

Done.

+   * that future, and can wait for it before continuing.
    *
    * 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(),
                             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 77 matching lines @@
    * allows.
    */
   void addError(Object error, [Object stackTrace]);
 }
 
 
 abstract class _StreamControllerLifecycle<T> {
   StreamSubscription<T> _subscribe(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 270 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);
+    Future future = _runGuarded(_onCancel);

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

If the doneFuture and the cancel are not linked (shouldn't they?)
move the _runGuarded below the completion of the _doneFuture.

[Anders Johnsen, 2013/10/21 08:01:46]

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;
 }
 
 class _NoCallbackAsyncStreamController/*<T>*/ = _StreamController/*<T>*/
        with _AsyncStreamControllerDispatch/*<T>*/, _NoCallbacks;
 
 class _NoCallbackSyncStreamController/*<T>*/ = _StreamController/*<T>*/
        with _SyncStreamControllerDispatch/*<T>*/, _NoCallbacks;
 
-typedef void _NotificationHandler();
+typedef _NotificationHandler();
 
-void _runGuarded(_NotificationHandler notificationHandler) {
-  if (notificationHandler == null) return;
+Future _runGuarded(_NotificationHandler notificationHandler) {
+  if (notificationHandler == null) return null;
   try {
-    notificationHandler();
+    var result = notificationHandler();
+    if (result is Future) return result;
+    return null;
   } catch (e, s) {
     Zone.current.handleUncaughtError(_asyncError(e, s), s);
   }
 }
 
 class _ControllerStream<T> extends _StreamImpl<T> {
   _StreamControllerLifecycle<T> _controller;
 
   _ControllerStream(this._controller);
 
@@ skipping 13 matching lines @@
     return identical(otherStream._controller, this._controller);
   }
 }
 
 class _ControllerSubscription<T> extends _BufferingStreamSubscription<T> {
   final _StreamControllerLifecycle<T> _controller;
 
   _ControllerSubscription(this._controller, bool cancelOnError)
       : super(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, [StackTrace stackTrace]) {
     _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();
     }
   }
 }