Return futures on Stream.cancel when possible.

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 170 matching lines @@
         }
       }
     }
   }
 
   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();
-    return _cancelFuture;
+    if (!_isCanceled) {
+      _cancel();
+    }
+    return _cancelFuture ?? Future._nullFuture;
   }
 
   Future/*<E>*/ asFuture/*<E>*/([var/*=E*/ futureValue]) {
     _Future/*<E>*/ result = new _Future/*<E>*/();
 
     // Overwrite the onDone and onError handlers.
     _onDone = () { result._complete(futureValue); };
     _onError = (error, stackTrace) {
-      cancel();
-      result._completeError(error, stackTrace);
+      Future cancelFuture = cancel();
+      if (cancelFuture is Future) {

[Lasse Reichstein Nielsen, 2016/08/02 01:25:39]

cancel() can't return null any more, so there is no need to check for `is
Future`. (If it's a non-null non-Future object, it's a type error, and it's fine
to just assume it's a Future).

Should the check be `!identical(cancelFuture, Future._nullFuture)`?

[floitsch, 2016/08/02 12:03:09]

done.

+        cancelFuture.then((_) {

[Lasse Reichstein Nielsen, 2016/08/02 01:25:39]

If the cancelFuture has an error, then the result future isn't completed. Maybe
use "whenComplete" instead of "then"?
The cancelFuture error will still be unhandled, but it's probably a good idea to
wait for it before completing the asFuture future.

[floitsch, 2016/08/02 12:03:09]

done.

+          result._completeError(error, stackTrace);
+        });
+      } else {
+        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;
@@ skipping 140 matching lines @@
       } else {
         _zone.runUnaryGuarded/*<dynamic, dynamic>*/(
             _onError as Object /*=ZoneUnaryCallback<dynamic, dynamic>*/, error);
       }
       _state &= ~_STATE_IN_CALLBACK;
     }
 
     if (_cancelOnError) {
       _state |= _STATE_WAIT_FOR_CANCEL;
       _cancel();
-      if (_cancelFuture is Future) {
+      if (_cancelFuture is Future &&
+          !identical(_cancelFuture, Future._nullFuture)) {
         _cancelFuture.whenComplete(sendError);
       } else {
         sendError();
       }
     } else {
       sendError();
       // Only check state if not cancelOnError.
       _checkState(wasInputPaused);
     }
   }
 
   void _sendDone() {
     assert(!_isCanceled);
     assert(!_isPaused);
     assert(!_inCallback);
 
     void sendDone() {
       // If the subscription has been canceled while waiting for the cancel
       // future to finish we must not report the done event.
       if (!_waitsForCancel) 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) {
+    if (_cancelFuture is Future &&
+        !identical(_cancelFuture, Future._nullFuture)) {
       _cancelFuture.whenComplete(sendDone);
     } else {
       sendDone();
     }
   }
 
   /**
    * Call a hook function.
    *
    * The call is properly wrapped in code to avoid other callbacks
@@ skipping 368 matching lines @@
 
   void resume() {
     if (isPaused) {
       _state -= _PAUSED;
       if (!isPaused && !_isSent) {
         _schedule();
       }
     }
   }
 
-  Future cancel() => null;
+  Future cancel() => Future._nullFuture;
 
   Future/*<E>*/ asFuture/*<E>*/([var/*=E*/ futureValue]) {
     _Future/*<E>*/ result = new _Future/*<E>*/();
     _onDone = () { result._completeWithValue(null); };
     return result;
   }
 
   void _sendDone() {
     _state &= ~_SCHEDULED;
     if (isPaused) return;
@@ skipping 117 matching lines @@
   void pause([Future resumeSignal]) {
     _stream._pauseSubscription(resumeSignal);
   }
 
   void resume() {
     _stream._resumeSubscription();
   }
 
   Future cancel() {
     _stream._cancelSubscription();
-    return null;
+    return Future._nullFuture;
   }
 
   bool get isPaused {
     return _stream._isSubscriptionPaused;
   }
 
   Future/*<E>*/ asFuture/*<E>*/([var/*=E*/ futureValue]) {
     throw new UnsupportedError(
         "Cannot change handlers of asBroadcastStream source subscription.");
   }
@@ skipping 95 matching lines @@
   /** Clears up the internal state when the iterator ends. */
   void _clear() {
     _subscription = null;
     _futureOrPrefetch = null;
     _current = null;
     _state = _STATE_DONE;
   }
 
   Future cancel() {
     StreamSubscription subscription = _subscription;
-    if (subscription == null) return null;
+    if (subscription == null) return Future._nullFuture;
     if (_state == _STATE_MOVING) {
       _Future<bool> hasNext = _futureOrPrefetch as Object /*=_Future<bool>*/;
       _clear();
       hasNext._complete(false);
     } else {
       _clear();
     }
     return subscription.cancel();
   }
 
@@ skipping 43 matching lines @@
 class _EmptyStream<T> extends Stream<T> {
   const _EmptyStream() : super._internal();
   bool get isBroadcast => true;
   StreamSubscription<T> listen(void onData(T data),
                                {Function onError,
                                 void onDone(),
                                 bool cancelOnError}) {
     return new _DoneStreamSubscription<T>(onDone);
   }
 }