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, StackTrace stackTrace);
@@ skipping 55 matching lines @@
    * after a call to [resume] if there are pending events.
    */
   static const int _STATE_INPUT_PAUSED = 4;
   /**
    * Whether the subscription has been canceled.
    *
    * Set by calling [cancel], or by handling a "done" event, or an "error" event
    * when `cancelOnError` is true.
    */
   static const int _STATE_CANCELED = 8;
-  static const int _STATE_IN_CALLBACK = 16;
-  static const int _STATE_HAS_PENDING = 32;
-  static const int _STATE_PAUSE_COUNT = 64;
-  static const int _STATE_PAUSE_COUNT_SHIFT = 6;
+  /**
+   * Set when either:
+   *   * An error is sent, and [cancelOnError] is true.

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

Nit (you also need the new line as otherwise dartdoc will not itemize).

Set when either:

  * an error is sent, and [cancelOnError] is true, or
  * a done event is sent.

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

Done.

+   *   * A done is sent.
+   *
+   * If the subscription is canceled while _STATE_WAIT_FOR_CANCEL is set, the
+   * state is unset, and no furher events should be delivered.

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

"should" is not helpful.
I guess "is" is more accurate (or "must").

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

Using 'must', as the comment is dictating how the implementation should act.

+   */
+  static const int _STATE_WAIT_FOR_CANCEL = 16;
+  static const int _STATE_IN_CALLBACK = 32;
+  static const int _STATE_HAS_PENDING = 64;
+  static const int _STATE_PAUSE_COUNT = 128;
+  static const int _STATE_PAUSE_COUNT_SHIFT = 7;
 
   /* Event handlers provided in constructor. */
   _DataHandler<T> _onData;
   Function _onError;
   _DoneHandler _onDone;
   final Zone _zone = Zone.current;
 
   /** Bit vector based on state-constants above. */
   int _state;
 
+  // TODO(floitsch): reuse another field
+  /** The future [_onCancel] may return. */
+  Future _cancelFuture;
+
   /**
    * Queue of pending events.
    *
    * Is created when necessary, or set in constructor for preconfigured events.
    */
   _PendingEvents _pending;
 
   _BufferingStreamSubscription(bool cancelOnError)
       : _state = (cancelOnError ? _STATE_CANCEL_ON_ERROR : 0);
 
@@ skipping 64 matching lines @@
           _pending.schedule(this);
         } else {
           assert(_mayResumeInput);
           _state &= ~_STATE_INPUT_PAUSED;
           if (!_inCallback) _guardCallback(_onResume);
         }
       }
     }
   }
 
-  void cancel() {
-    if (_isCanceled) return;
+  Future cancel() {
+    // The user doesn't want to receive any further events. If there is an
+    // error or done event pending (waiting for the cancel to be done) discard
+    // that event.
+    _state &= ~_STATE_WAIT_FOR_CANCEL;
+    if (_isCanceled) return _cancelFuture;
     _cancel();
-    if (!_inCallback) {
-      // otherwise checkState will be called after firing or callback completes.
-      _state |= _STATE_IN_CALLBACK;
-      _onCancel();
-      _pending = null;
-      _state &= ~_STATE_IN_CALLBACK;
-    }
+    return _cancelFuture;
   }
 
   Future asFuture([var futureValue]) {
     _Future<T> result = new _Future<T>();
 
     // Overwrite the onDone and onError handlers.
     _onDone = () { result._complete(futureValue); };
     _onError = (error, stackTrace) {
       cancel();
       result._completeError(error, stackTrace);
     };
 
     return result;
   }
 
   // State management.
 
   bool get _isInputPaused => (_state & _STATE_INPUT_PAUSED) != 0;
   bool get _isClosed => (_state & _STATE_CLOSED) != 0;
   bool get _isCanceled => (_state & _STATE_CANCELED) != 0;
+  bool get _waitForCancel => (_state & _STATE_WAIT_FOR_CANCEL) != 0;

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

_waitsForCancel

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

Done.

   bool get _inCallback => (_state & _STATE_IN_CALLBACK) != 0;
   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() {
     _state |= _STATE_CANCELED;
     if (_hasPending) {
       _pending.cancelSchedule();
     }
+    if (!_inCallback) _pending = null;
+    _cancelFuture = _onCancel();
   }
 
   /**
    * Increment the pause count.
    *
    * Also marks input as paused.
    */
   void _incrementPauseCount() {
     _state = (_state + _STATE_PAUSE_COUNT) | _STATE_INPUT_PAUSED;
   }
@@ skipping 47 matching lines @@
   // try/catch wrapping and send any errors to
   // [_Zone.current.handleUncaughtError].
   void _onPause() {
     assert(_isInputPaused);
   }
 
   void _onResume() {
     assert(!_isInputPaused);
   }
 
-  void _onCancel() {
+  Future _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 21 matching lines @@
     _zone.runUnaryGuarded(_onData, data);
     _state &= ~_STATE_IN_CALLBACK;
     _checkState(wasInputPaused);
   }
 
   void _sendError(var error, StackTrace stackTrace) {
     assert(!_isCanceled);
     assert(!_isPaused);
     assert(!_inCallback);
     bool wasInputPaused = _isInputPaused;
-    _state |= _STATE_IN_CALLBACK;
-    if (!_zone.inSameErrorZone(Zone.current)) {
-      // Errors are not allowed to traverse zone boundaries.
-      Zone.current.handleUncaughtError(error, stackTrace);
-    } else if (_onError is ZoneBinaryCallback) {
-      _zone.runBinaryGuarded(_onError, error, stackTrace);
+
+    void sendError() {
+      // If the subscription has been canceled while waiting for the cancel
+      // future to finish we must not report the error.
+      if (_isCanceled && !_waitForCancel) return;
+      _state |= _STATE_IN_CALLBACK;
+      if (!_zone.inSameErrorZone(Zone.current)) {
+        // Errors are not allowed to traverse zone boundaries.
+        Zone.current.handleUncaughtError(error, stackTrace);
+      } else if (_onError is ZoneBinaryCallback) {
+        _zone.runBinaryGuarded(_onError, error, stackTrace);
+      } else {
+        _zone.runUnaryGuarded(_onError, error);
+      }
+      _state &= ~_STATE_IN_CALLBACK;
+    }
+
+    if (_cancelOnError) {
+      _state |= _STATE_WAIT_FOR_CANCEL;
+      _cancel();
+      if (_cancelFuture is Future) {
+        _cancelFuture.whenComplete(sendError);
+      } else {
+        sendError();
+      }
     } else {
-      _zone.runUnaryGuarded(_onError, error);
+      sendError();
+      // Only check state if not cancelOnError.
+      _checkState(wasInputPaused);
     }
-    _state &= ~_STATE_IN_CALLBACK;
-    if (_cancelOnError) {
-      _cancel();
-    }
-    _checkState(wasInputPaused);
   }
 
   void _sendDone() {
     assert(!_isCanceled);
     assert(!_isPaused);
     assert(!_inCallback);
-    _state |= (_STATE_CANCELED | _STATE_CLOSED | _STATE_IN_CALLBACK);
-    _zone.runGuarded(_onDone);
-    _onCancel();  // No checkState after cancel, it is always the last event.
-    _state &= ~_STATE_IN_CALLBACK;
+
+    void sendDone() {
+      // If the subscription has been canceled while waiting for the cancel
+      // future to finish we must not report the done event.
+      if (!_waitForCancel) return;
+      _state |= (_STATE_CANCELED | _STATE_CLOSED | _STATE_IN_CALLBACK);
+      _zone.runGuarded(_onDone);
+      _state &= ~_STATE_IN_CALLBACK;
+    }
+
+    _cancel();
+    _state |= _STATE_WAIT_FOR_CANCEL;
+    if (_cancelFuture is Future) {
+      _cancelFuture.whenComplete(sendDone);
+    } else {
+      sendDone();
+    }
   }
 
   /**
    * Call a hook function.
    *
    * The call is properly wrapped in code to avoid other callbacks
    * during the call, and it checks for state changes after the call
    * that should cause further callbacks.
    */
   void _guardCallback(callback) {
@@ skipping 20 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();
         _pending = null;
         return;
       }
       bool isInputPaused = _isInputPaused;
       if (wasInputPaused == isInputPaused) break;
       _state ^= _STATE_IN_CALLBACK;
       if (isInputPaused) {
         _onPause();
       } else {
         _onResume();
@@ skipping 289 matching lines @@
   void onError(Function handleError) {}
   void onDone(void handleDone()) {}
 
   void pause([Future resumeSignal]) {
     _pauseCounter++;
     if (resumeSignal != null) resumeSignal.then((_) { resume(); });
   }
   void resume() {
     if (_pauseCounter > 0) _pauseCounter--;
   }
-  void cancel() {}
+  Future cancel() => null;
   bool get isPaused => _pauseCounter > 0;
 
   Future asFuture([futureValue]) => new _Future();
 }
 
 class _AsBroadcastStream<T> extends Stream<T> {
   final Stream<T> _source;
   final _broadcastCallback _onListenHandler;
   final _broadcastCallback _onCancelHandler;
   final Zone _zone;
@@ skipping 103 matching lines @@
   }
 
   void pause([Future resumeSignal]) {
     _stream._pauseSubscription(resumeSignal);
   }
 
   void resume() {
     _stream._resumeSubscription();
   }
 
-  void cancel() {
+  Future cancel() {
     _stream._cancelSubscription();
+    return null;
   }
 
   bool get isPaused {
     return _stream._isSubscriptionPaused;
   }
 
   Future asFuture([var futureValue]) {
     throw new UnsupportedError(
         "Cannot change handlers of asBroadcastStream source subscription.");
   }
@@ skipping 90 matching lines @@
   }
 
   /** Clears up the internal state when the iterator ends. */
   void _clear() {
     _subscription = null;
     _futureOrPrefetch = null;
     _current = null;
     _state = _STATE_DONE;
   }
 
-  void cancel() {
+  Future cancel() {
     StreamSubscription subscription = _subscription;
     if (_state == _STATE_MOVING) {
       _Future<bool> hasNext = _futureOrPrefetch;
       _clear();
       hasNext._complete(false);
     } else {
       _clear();
     }
-    subscription.cancel();
+    return subscription.cancel();
   }
 
   void _onData(T data) {
     if (_state == _STATE_MOVING) {
       _current = data;
       _Future<bool> hasNext = _futureOrPrefetch;
       _futureOrPrefetch = null;
       _state = _STATE_FOUND;
       hasNext._complete(true);
       return;
@@ skipping 23 matching lines @@
       _Future<bool> hasNext = _futureOrPrefetch;
       _clear();
       hasNext._complete(false);
       return;
     }
     _subscription.pause();
     _futureOrPrefetch = null;
     _state = _STATE_EXTRA_DONE;
   }
 }