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 495d04790ae894487d9df70dfdd48da98fda83fd..cdeb7b741b6c9c058d956b29a1f2a89633ba0d1e 100644
--- a/sdk/lib/async/stream.dart
+++ b/sdk/lib/async/stream.dart
@@ -347,8 +347,7 @@ abstract class Stream<T> {
         try {
           buffer.write(element);
         } catch (e, s) {
-          subscription.cancel();
-          result._completeError(_asyncError(e, s));
+          _cancelAndError(subscription, result)(_asyncError(e, s));

[floitsch, 2013/10/12 18:53:57]

It seems wrong to create a closure and then invoke it immediately...

[Lasse Reichstein Nielsen, 2013/10/14 11:32:33]

Agree.
If it is because _cancelAndError(subscription, result) is used somewhere else as
an error-callback, I'd rather have three arguments and use
   (x) => _cancelAndError(subscription, result, x) 
at that point. Which is already happening (check the (T value) => ...) change
below).

[Anders Johnsen, 2013/10/16 11:52:21]

Done.

[Anders Johnsen, 2013/10/16 11:52:21]

Done.

         }
       },
       onError: (e) {
@@ -376,8 +375,7 @@ abstract class Stream<T> {
             () => (element == needle),
             (bool isMatch) {
               if (isMatch) {
-                subscription.cancel();
-                future._complete(true);
+                _cancelAndValue(subscription, future)(true);
               }
             },
             _cancelAndError(subscription, future)
@@ -432,8 +430,7 @@ abstract class Stream<T> {
             () => test(element),
             (bool isMatch) {
               if (!isMatch) {
-                subscription.cancel();
-                future._complete(false);
+                _cancelAndValue(subscription, future)(false);
               }
             },
             _cancelAndError(subscription, future)
@@ -462,8 +459,7 @@ abstract class Stream<T> {
             () => test(element),
             (bool isMatch) {
               if (isMatch) {
-                subscription.cancel();
-                future._complete(true);
+                _cancelAndValue(subscription, future)(true);
               }
             },
             _cancelAndError(subscription, future)
@@ -498,8 +494,7 @@ abstract class Stream<T> {
     StreamSubscription subscription;
     subscription = this.listen(
       (_) {
-        subscription.cancel();
-        future._complete(false);
+        _cancelAndValue(subscription, future)(false);
       },
       onError: future._completeError,
       onDone: () {
@@ -631,9 +626,7 @@ abstract class Stream<T> {
     StreamSubscription subscription;
     subscription = this.listen(
       (T value) {
-        subscription.cancel();
-        future._complete(value);
-        return;
+        _cancelAndValue(subscription, future)(value);
       },
       onError: future._completeError,
       onDone: () {
@@ -687,10 +680,9 @@ 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._completeError(error);
+          _cancelAndError(subscription, future)(error);
           return;
         }
         foundResult = true;
@@ -731,8 +723,7 @@ abstract class Stream<T> {
           () => test(value),
           (bool isMatch) {
             if (isMatch) {
-              subscription.cancel();
-              future._complete(value);
+              _cancelAndValue(subscription, future)(value);
             }
           },
           _cancelAndError(subscription, future)
@@ -809,8 +800,7 @@ abstract class Stream<T> {
           (bool isMatch) {
             if (isMatch) {
               if (foundResult) {
-                subscription.cancel();
-                future._completeError(
+                _cancelAndError(subscription, future)(
                     new StateError('Multiple matches for "single"'));
                 return;
               }
@@ -851,8 +841,7 @@ abstract class Stream<T> {
     subscription = this.listen(
       (T value) {
         if (index == 0) {
-          subscription.cancel();
-          future._complete(value);
+          _cancelAndValue(subscription, future)(value);
           return;
         }
         index -= 1;
@@ -880,8 +869,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
+   * wait for the operation to complete. If in doubt, wait for it.

[floitsch, 2013/10/12 18:53:57]

Note that it can be `null`.

[Anders Johnsen, 2013/10/16 11:52:21]

Done.

    */
-  void cancel();
+  Future cancel();
 
   /** Set or override the data event handler of this subscription. */
   void onData(void handleData(T data));
@@ -1279,5 +1271,5 @@ abstract class StreamIterator<T> {
    * automatically closed, you must call [cancel] to ensure that the stream
    * is properly closed.

[floitsch, 2013/10/12 18:53:57]

Add comment (even if it is just pointing to the StreamSubscription).
Preferably without link.

[Anders Johnsen, 2013/10/16 11:52:21]

Done.

    */
-  void cancel();
+  Future cancel();
 }