Let StreamSubscription.cancel return a Future.

sdk/lib/async/stream.dart

Index: sdk/lib/async/stream.dart
diff --git a/sdk/lib/async/stream.dart b/sdk/lib/async/stream.dart
index 222ad8d37494c5a1aded6c6e8d87ed004f4b00df..573f80e8bfe7416e8f92cb9c9a94352fd5944eca 100644
--- a/sdk/lib/async/stream.dart
+++ b/sdk/lib/async/stream.dart
@@ -347,8 +347,9 @@ abstract class Stream<T> {
         try {
           buffer.write(element);
         } catch (e, s) {
-          subscription.cancel();
-          result._setError(_asyncError(e, s));
+          subscription.cancel().whenComplete(() {

[floitsch, 2013/07/12 16:42:34]

Why `whenComplete` if `cancel` is guaranteed not to throw an error?
`whenComplete` might make the closure nicer to write, but it also costs more
(since it needs to register itself for both callbacks).

[Lasse Reichstein Nielsen, 2013/07/16 12:03:31]

Why would that cost more?

[floitsch, 2013/07/16 12:47:17]

My mistake.

+            result._setError(_asyncError(e, s));
+          });
         }
       },
       onError: (e) {
@@ -376,8 +377,9 @@ abstract class Stream<T> {
             () => (element == needle),
             (bool isMatch) {
               if (isMatch) {
-                subscription.cancel();
-                future._setValue(true);
+                subscription.cancel().whenComplete(() {
+                  future._setValue(true);
+                });
               }
             },
             _cancelAndError(subscription, future)
@@ -432,8 +434,9 @@ abstract class Stream<T> {
             () => test(element),
             (bool isMatch) {
               if (!isMatch) {
-                subscription.cancel();
-                future._setValue(false);
+                subscription.cancel().whenComplete(() {
+                  future._setValue(false);
+                });
               }
             },
             _cancelAndError(subscription, future)
@@ -462,8 +465,9 @@ abstract class Stream<T> {
             () => test(element),
             (bool isMatch) {
               if (isMatch) {
-                subscription.cancel();
-                future._setValue(true);
+                subscription.cancel().whenComplete(() {
+                  future._setValue(true);
+                });
               }
             },
             _cancelAndError(subscription, future)
@@ -498,8 +502,9 @@ abstract class Stream<T> {
     StreamSubscription subscription;
     subscription = this.listen(
       (_) {
-        subscription.cancel();
-        future._setValue(false);
+        subscription.cancel().whenComplete(() {
+          future._setValue(false);
+        });
       },
       onError: future._setError,
       onDone: () {
@@ -623,9 +628,9 @@ abstract class Stream<T> {
     StreamSubscription subscription;
     subscription = this.listen(
       (T value) {
-        subscription.cancel();
-        future._setValue(value);
-        return;
+        subscription.cancel().whenComplete(() {
+          future._setValue(value);
+        });
       },
       onError: future._setError,
       onDone: () {
@@ -675,10 +680,11 @@ abstract class Stream<T> {
     subscription = this.listen(
       (T value) {
         if (foundResult) {
-          subscription.cancel();
-          // This is the second element we get.
-          Error error = new StateError("More than one element");
-          future._setError(error);
+          subscription.cancel().whenComplete(() {
+            // This is the second element we get.
+            Error error = new StateError("More than one element");
+            future._setError(error);
+          });
           return;
         }
         foundResult = true;
@@ -719,8 +725,9 @@ abstract class Stream<T> {
           () => test(value),
           (bool isMatch) {
             if (isMatch) {
-              subscription.cancel();
-              future._setValue(value);
+              subscription.cancel().whenComplete(() {
+                future._setValue(value);
+              });
             }
           },
           _cancelAndError(subscription, future)
@@ -797,9 +804,10 @@ abstract class Stream<T> {
           (bool isMatch) {
             if (isMatch) {
               if (foundResult) {
-                subscription.cancel();
-                future._setError(
-                    new StateError('Multiple matches for "single"'));
+                subscription.cancel().whenComplete(() {
+                  future._setError(
+                      new StateError('Multiple matches for "single"'));
+                });
                 return;
               }
               foundResult = true;
@@ -836,8 +844,9 @@ abstract class Stream<T> {
     subscription = this.listen(
       (T value) {
         if (index == 0) {
-          subscription.cancel();
-          future._setValue(value);
+          subscription.cancel().whenComplete(() {
+            future._setValue(value);
+          });
           return;
         }
         index -= 1;
@@ -865,8 +874,11 @@ abstract class StreamSubscription<T> {
    *
    * If an event is currently firing, this unsubscription will only
    * take effect after all subscribers have received the current event.
+   *
+   * In case the cancel is asynchronous, the returned [Future] can be used to

[floitsch, 2013/07/12 16:42:34]

Returns a future that is completed when the cancel is done. The returned future
is guaranteed not to complete with an error.

[Lasse Reichstein Nielsen, 2013/07/17 07:28:46]

Agree, if it always returns a future, then there is no need to say "in case",
because it's impossible to say if it is the case and you should always act as if
it was async.

+   * wait for the operation to complete. If in doubt, wait for it.
    */
-  void cancel();
+  Future cancel();
 
   /** Set or override the data event handler of this subscription. */
   void onData(void handleData(T data));
@@ -1264,5 +1276,5 @@ abstract class StreamIterator<T> {
    * automatically closed, you must call [cancel] to ensure that the stream
    * is properly closed.
    */
-  void cancel();
+  Future cancel();
 }