Return futures on Stream.cancel when possible.

sdk/lib/async/stream_pipe.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;
 
 /** Runs user code and takes actions depending on success or failure. */
 _runUserCode(userCode(),
              onSuccess(value),
              onError(error, StackTrace stackTrace)) {
@@ skipping 11 matching lines @@
   }
 }
 
 /** Helper function to cancel a subscription and wait for the potential future,
   before completing with an error. */
 void _cancelAndError(StreamSubscription subscription,
                      _Future future,
                      error,
                      StackTrace stackTrace) {
   var cancelFuture = subscription.cancel();
-  if (cancelFuture is Future) {
+  if (cancelFuture is Future && !identical(cancelFuture, Future._nullFuture)) {
     cancelFuture.whenComplete(() => future._completeError(error, stackTrace));
   } else {
     future._completeError(error, stackTrace);
   }
 }
 
 void _cancelAndErrorWithReplacement(StreamSubscription subscription,
                                     _Future future,
                                     error, StackTrace stackTrace) {
   AsyncError replacement = Zone.current.errorCallback(error, stackTrace);
@@ skipping 11 matching lines @@
     StreamSubscription subscription, _Future future) {
   return (error, StackTrace stackTrace) {
     _cancelAndError(subscription, future, error, stackTrace);
   };
 }
 
 /** Helper function to cancel a subscription and wait for the potential future,
   before completing with a value. */
 void _cancelAndValue(StreamSubscription subscription, _Future future, value) {
   var cancelFuture = subscription.cancel();
-  if (cancelFuture is Future) {
+  if (cancelFuture is Future && !identical(cancelFuture, Future._nullFuture)) {
     cancelFuture.whenComplete(() => future._complete(value));
   } else {
     future._complete(value);
   }
 }
 
 
 /**
  * A [Stream] that forwards subscriptions to another stream.
  *
@@ skipping 411 matching lines @@
         _addErrorWithReplacement(sink, e, s);
         return null;
       }
       if (!isEqual) {
         sink._add(inputEvent);
         _previous = inputEvent;
       }
     }
   }
 }