Let StreamSubscription.cancel return a Future.

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;
 
 /**
  * Utility function to attach a stack trace to an [error]  if it doesn't have
  * one already.
  */
 _asyncError(Object error, Object stackTrace) {
   if (stackTrace == null) return error;
   if (getAttachedStackTrace(error) != null) return error;
   _attachStackTrace(error, stackTrace);
   return error;
 }
 
 /** Runs user code and takes actions depending on success or failure. */
 _runUserCode(userCode(),
              onSuccess(value),
              onError(error, StackTrace stackTrace)) {
   try {
     onSuccess(userCode());
   } catch (e, s) {
     onError(_asyncError(e, s), s);
   }
 }
 
+/** 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) {
+    cancelFuture.whenComplete(() => future._completeError(error, stackTrace));
+  } else {
+    future._completeError(error, stackTrace);
+  }
+}
+
 /** Helper function to make an onError argument to [_runUserCode]. */
-_cancelAndError(StreamSubscription subscription, _Future future) =>
-  (error, StackTrace stackTrace) {
-    subscription.cancel();
-    future._completeError(error, stackTrace);
-  };
+_cancelAndErrorClosure(StreamSubscription subscription, _Future future) =>
+  ((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) {
+    cancelFuture.whenComplete(() => future._complete(value));
+  } else {
+    future._complete(value);
+  }
+}
 
 
 /**
  * A [Stream] that forwards subscriptions to another stream.
  *
  * This stream implements [Stream], but forwards all subscriptions
  * to an underlying stream, and wraps the returned subscription to
  * modify the events on the way.
  *
  * This class is intended for internal use only.
@@ skipping 340 matching lines @@
         sink._addError(_asyncError(e, s), s);
         return null;
       }
       if (!isEqual) {
         sink._add(inputEvent);
         _previous = inputEvent;
       }
     }
   }
 }