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.

[Lasse Reichstein Nielsen, 2013/07/17 07:28:46]

The [onCancel] callback may return a value or [Future]. The
[StreamSubscription.cancel] call returns a [Future] that will be completed with
this value, or with the value of the [Future].
A [Future] returned by [onCancel] should not complete with an error - the error
will most likely be unhandled. Also, any value is likely to be ignored, so just
returning or completing with a null value is sufficient.

    *
    * 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 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) {}
+  _FutureImpl _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) {
+  _FutureImpl _recordCancel(StreamSubscription<T> subscription) {

[Lasse Reichstein Nielsen, 2013/07/17 07:28:46]

Why _FutureImpl if the user can provide a future through the StreamController?
If this was just Future, then we could forward the user's Future directly, and
only wrap it if it is not a future.

     if (_isAddingStream) {
       _StreamControllerAddStreamState addState = _varData;
       addState.cancel();
     }
     _varData = null;
     _state =
         (_state & ~(_STATE_SUBSCRIBED | _STATE_ADDSTREAM)) | _STATE_CANCELED;
-    _runGuarded(_onCancel);
+    var future = _runGuardedFutureResult(_onCancel);

[floitsch, 2013/07/12 16:42:34]

_runGuardedFutureResult may return `null`. Document that this is ok (if it is).

I would still call _runGuarded and do the FutureImpl test here.
(unless _runGuardedFutureResult is supposed to be invoked at other locations).

If _runGuardedFutureResult receives a future, you don't verify that it doesn't
complete with an error. However we guarantee that the future never completes
with an error.

[Lasse Reichstein Nielsen, 2013/07/17 07:28:46]

I think we should not guarantee that there is no errors when the user is allowed
to provide the error - which they are through the StreamController.

We can recommend that there is no error, and say that if there is, it won't be
handled, but in the end, it's the user that decides how to complete.

     if (_doneFuture != null && _doneFuture._mayComplete) {
       _doneFuture._asyncSetValue(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;
   try {
     notificationHandler();
   } catch (e, s) {
     _Zone.current.handleUncaughtError(_asyncError(e, s));
   }
 }
 
+_FutureImpl _runGuardedFutureResult(_NotificationHandler notificationHandler) {
+  if (notificationHandler == null) return null;
+  try {
+    var value = notificationHandler();
+    if (value is _FutureImpl) {
+      return value;
+    } else {

[Lasse Reichstein Nielsen, 2013/07/17 07:28:46]

else if (value is Future) {  // You can't assume that all futures are
_FutureImpl.
  Future future = value;
  _FutureImpl result = new _FutureImpl();
  future.then(result._setValue, onError: result._setError);
  return future;
} else {

+      return new Future(() => value);

[Lasse Reichstein Nielsen, 2013/07/17 07:28:46]

return new _FutureImpl.immediate(value);

+    }
+  } catch (e, s) {
+    _Zone.current.handleUncaughtError(_asyncError(e, s));
+  }
+}
+
 class _ControllerStream<T> extends _StreamImpl<T> {
   _StreamControllerLifecycle<T> _controller;
 
   _ControllerStream(this._controller);
 
   StreamSubscription<T> _createSubscription(
       void onData(T data),
       void onError(Object error),
       void onDone(),
       bool cancelOnError) =>
@@ skipping 15 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);
+  _FutureImpl _onCancel() {
+    return _controller._recordCancel(this);
   }
 
   void _onPause() {
     _controller._recordPause(this);
   }
 
   void _onResume() {
     _controller._recordResume(this);
   }
 }
@@ skipping 52 matching lines @@
   var varData;
 
   _StreamControllerAddStreamState(_StreamController controller,
                                   this.varData,
                                   Stream source) : super(controller, source) {
     if (controller.isPaused) {
       addSubscription.pause();
     }
   }
 }