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 53 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;
+  static const int _STATE_IN_ERROR_CANCEL = 16;

[floitsch, 2013/10/12 18:53:57]

Add comment that _STATE_IN_ERROR_CANCEL implies that the _STATE_CANCEL bit is
set, too.

[Lasse Reichstein Nielsen, 2013/10/14 11:32:33]

And document what it means. In painful detail, because this is so hard to
understand anyway.

[Anders Johnsen, 2013/10/16 11:52:21]

Done.

+  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;
   _ErrorHandler _onError;
   _DoneHandler _onDone;
   final Zone _zone = Zone.current;
 
   /** Bit vector based on state-constants above. */
   int _state;
 
+  _Future _cancelFuture;

[floitsch, 2013/10/12 18:53:57]

Add "TODO(floitsch): reuse another field"

[Anders Johnsen, 2013/10/16 11:52:21]

Done.

+
   /**
    * Queue of pending events.
    *
    * Is created when necessary, or set in constructor for preconfigured events.
    */
   _PendingEvents _pending;
 
   _BufferingStreamSubscription(void onData(T data),
                                void onError(error),
                                void onDone(),
@@ skipping 74 matching lines @@
           _pending.schedule(this);
         } else {
           assert(_mayResumeInput);
           _state &= ~_STATE_INPUT_PAUSED;
           if (!_inCallback) _guardCallback(_onResume);
         }
       }
     }
   }
 
-  void cancel() {
-    if (_isCanceled) return;
+  Future cancel() {
+    _state &= ~_STATE_IN_ERROR_CANCEL;

[floitsch, 2013/10/12 18:53:57]

Add comment.
// The user doesn't want to receive any events anymore. If there is an
// error event pending (waiting for the cancel to be done) discard that
// error.

[Lasse Reichstein Nielsen, 2013/10/14 11:32:33]

any events anymore -> any further events.

This could possibly be generalized to handle the pending done event as well. If
the user cancels before an event, which has been delayed because it waits for
cancelling to complete, that event is no longer needed.

[Anders Johnsen, 2013/10/16 11:52:21]

Done.

[Anders Johnsen, 2013/10/16 11:52:21]

Done.

+    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) {
       cancel();
       result._completeError(error);
     };
 
     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 _inErrorCancel => (_state & _STATE_IN_ERROR_CANCEL) != 0;
   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();

[floitsch, 2013/10/12 18:53:57]

I'm not sure we are allowed to call "_onCancel" while there is a callback
executed.
The original code guarded against this. Please talk to Lasse.

[Lasse Reichstein Nielsen, 2013/10/14 11:32:33]

I accepted that we had to call _onCancel during another callback.
The alternative was to return a new _Future and only later, during the _onCancel
callback, link the cancel future to the new future. But that would mean always
returning a future, even if _onCancel returned null.

The callback is likely be in an event handler, not in, e.g., an "onPause"
callback.

I'm not absolutely convinced that it's sound, but it looks like it.

[Anders Johnsen, 2013/10/16 11:52:21]

Done.

[Anders Johnsen, 2013/10/16 11:52:21]

Done.

   }
 
   /**
    * 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) {
     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);
+    void sendError() {

[floitsch, 2013/10/12 18:53:57]

one line before and one line after the function.

[Anders Johnsen, 2013/10/16 11:52:21]

Done.

+      if (!_isCanceled || _inErrorCancel) {

[floitsch, 2013/10/12 18:53:57]

Comment: If the subscription has been canceled while waiting for the cancel
future to finish we must not report the error.
Rewrite it for an early exit:
if (isCanceled && !_inErrorCancel) return;

[Anders Johnsen, 2013/10/16 11:52:21]

Done.

+        _state |= _STATE_IN_CALLBACK;
+        if (!_zone.inSameErrorZone(Zone.current)) {
+          // Errors are not allowed to traverse zone boundaries.
+          Zone.current.handleUncaughtError(error);
+        } else {
+          _zone.runUnaryGuarded(_onError, error);
+        }
+        _state &= ~_STATE_IN_CALLBACK;
+      }
+    }
+    if (_cancelOnError) {
+      _state |= _STATE_IN_ERROR_CANCEL;
+      _cancel();
+      if (_cancelFuture != null) {
+        _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.
+    // TODO(ajohnsen): Run it before _onDone and wait for future?

[Anders Johnsen, 2013/10/11 12:32:40]

Please comment on this one - I'm not sure what the right thing is to do (but I'm
leaning toward waiting for it).

[floitsch, 2013/10/12 18:53:57]

Neither.
The done event should *not* call cancel. Done implies that nothing comes
anymore.
The "_onCancel" is a misnomer. It should be "_onFinish" (or something similar).

It's the StreamController that should call its onCancel (which again is a
misnomer there). We should discuss this again in person.

[Anders Johnsen, 2013/10/16 11:52:21]

Done.

[Anders Johnsen, 2013/10/16 11:52:21]

Done.

+    _cancel();  // No checkState after cancel, it is always the last event.
     _state &= ~_STATE_IN_CALLBACK;
   }
 
   /**
    * 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.
    */
@@ skipping 21 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 298 matching lines @@
   void onError(void handleError(Object data)) {}
   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 102 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;
   }
 }