Return futures on Stream.cancel when possible.

sdk/lib/async/stream_pipe.dart

Index: sdk/lib/async/stream_pipe.dart
diff --git a/sdk/lib/async/stream_pipe.dart b/sdk/lib/async/stream_pipe.dart
index 1125620aacb3b7b445c3745e8a079c725a980f37..d26b0c5f64c86fc97c1df3c06a879e81c9c480fd 100644
--- a/sdk/lib/async/stream_pipe.dart
+++ b/sdk/lib/async/stream_pipe.dart
@@ -29,7 +29,7 @@ void _cancelAndError(StreamSubscription subscription,
                      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);
@@ -61,7 +61,7 @@ _ErrorCallback _cancelAndErrorClosure(
   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);