Let StreamSubscription.cancel return a Future.

sdk/lib/async/stream_impl.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;
 
 /** Abstract and private interface for a place to put events. */
 abstract class _EventSink<T> {
   void _add(T data);
   void _addError(Object error);
@@ skipping 67 matching lines @@
 
   /* Event handlers provided in constructor. */
   _DataHandler<T> _onData;
   _ErrorHandler _onError;
   _DoneHandler _onDone;
   final _Zone _zone = _Zone.current;
 
   /** Bit vector based on state-constants above. */
   int _state;
 
+  _FutureImpl _cancelFuture;

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

Add TODO that this should be in some field that isn't used anymore.

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

And add documentation saying what it is, and where/when it is being useed.

+
   /**
    * Queue of pending events.
    *
    * Is created when necessary, or set in constructor for preconfigured events.
    */
   _PendingEvents _pending;
 
   _BufferingStreamSubscription(this._onData,
                                this._onError,
                                this._onDone,
@@ skipping 72 matching lines @@
           _pending.schedule(this);
         } else {
           assert(_mayResumeInput);
           _state &= ~_STATE_INPUT_PAUSED;
           if (!_inCallback) _guardCallback(_onResume);
         }
       }
     }
   }
 
-  void cancel() {
-    if (_isCanceled) return;
+  void _chainCancelFuture(_FutureImpl future) {

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

_cancelFuture._setOrChainValue(future);

It handles any value (all Futures are chained, non-futures are set as values).

+    if (future == null) {
+      _cancelFuture._setValue(null);
+    } else {
+      future._chain(_cancelFuture);
+    }
+  }
+
+  Future cancel() {
+    if (_isCanceled) return _cancelFuture;
     _cancel();
     if (!_inCallback) {
       // otherwise checkState will be called after firing or callback completes.
       _state |= _STATE_IN_CALLBACK;
-      _onCancel();
+      _chainCancelFuture(_onCancel());

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

So this line could just be:
 _cancelFuture.setOrChainValue(onCancel());
and you would also handle non-_FutureImpl futures correctly.

       _pending = null;
       _state &= ~_STATE_IN_CALLBACK;
     }
+    return _cancelFuture;
   }
 
   Future asFuture([var futureValue]) {
     _FutureImpl<T> result = new _FutureImpl<T>();
 
     // Overwrite the onDone and onError handlers.
     _onDone = () { result._setValue(futureValue); };
     _onError = (error) {
       cancel();
       result._setError(error);
@@ skipping 11 matching lines @@
   bool get _hasPending => (_state & _STATE_HAS_PENDING) != 0;
   bool get _isPaused => _state >= _STATE_PAUSE_COUNT;
   bool get _canFire => _state < _STATE_IN_CALLBACK;
   bool get _mayResumeInput =>
       !_isPaused && (_pending == null || _pending.isEmpty);
   bool get _cancelOnError => (_state & _STATE_CANCEL_ON_ERROR) != 0;
 
   bool get isPaused => _isPaused;
 
   void _cancel() {
+    _cancelFuture = new _FutureImpl();
     _state |= _STATE_CANCELED;
     _zone.cancelCallbackExpectation();
     if (_hasPending) {
       _pending.cancelSchedule();
     }
   }
 
   /**
    * Increment the pause count.
    *
@@ skipping 52 matching lines @@
   // try/catch wrapping and send any errors to
   // [_Zone.current.handleUncaughtError].
   void _onPause() {
     assert(_isInputPaused);
   }
 
   void _onResume() {
     assert(!_isInputPaused);
   }
 
-  void _onCancel() {
+  _FutureImpl _onCancel() {
     assert(_isCanceled);
   }
 
   // Handle pending events.
 
   /**
    * Add a pending event.
    *
    * If the subscription is not paused, this also schedules a firing
    * of pending events later (if necessary).
@@ skipping 83 matching lines @@
     if (_hasPending && _pending.isEmpty) {
       _state &= ~_STATE_HAS_PENDING;
       if (_isInputPaused && _mayResumeInput) {
         _state &= ~_STATE_INPUT_PAUSED;
       }
     }
     // If the state changes during a callback, we immediately
     // make a new state-change callback. Loop until the state didn't change.
     while (true) {
       if (_isCanceled) {
-        _onCancel();
+        _chainCancelFuture(_onCancel());

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

_cancelFuture._setOrChainValue(_onCancel());

         _pending = null;
         return;
       }
       bool isInputPaused = _isInputPaused;
       if (wasInputPaused == isInputPaused) break;
       _state ^= _STATE_IN_CALLBACK;
       if (isInputPaused) {
         _onPause();
       } else {
         _onResume();
@@ skipping 592 matching lines @@
       _FutureImpl<bool> hasNext = _futureOrPrefetch;
       _clear();
       hasNext._setValue(false);
       return;
     }
     _subscription.pause();
     _futureOrPrefetch = null;
     _state = _STATE_EXTRA_DONE;
   }
 }