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 760 matching lines @@
 
   void resume() {
     if (isPaused) {
       _state -= _PAUSED;
       if (!isPaused && !_isSent) {
         _schedule();
       }
     }
   }
 
-  Future cancel() => null;
+  Future cancel() => new Future.value(null);

[Lasse Reichstein Nielsen, 2016/08/01 15:23:26]

Future._nullFuture

[floitsch, 2016/08/01 21:00:37]

Done.

 
   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 new Future.value(null);

[Lasse Reichstein Nielsen, 2016/08/01 15:23:26]

Future._nullFuture
(I will stop saying it now, but do change the rest too :)

[floitsch, 2016/08/01 21:00:37]

Done.

   }
 
   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 new Future.value(null);
     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);
   }
 }