Let StreamSubscription.cancel return a Future.

sdk/lib/async/stream.dart

 // Copyright (c) 2013, 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;
 
 // -------------------------------------------------------------------
 // Core Stream types
 // -------------------------------------------------------------------
 
@@ skipping 329 matching lines @@
     bool first = true;
     subscription = this.listen(
       (T element) {
         if (!first) {
           buffer.write(separator);
         }
         first = false;
         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) {
         result._setError(e);
       },
       onDone: () {
         result._setValue(buffer.toString());
       },
       cancelOnError: true);
     return result;
   }
 
   /**
    * Checks whether [needle] occurs in the elements provided by this stream.
    *
    * Completes the [Future] when the answer is known.
    * If this stream reports an error, the [Future] will report that error.
    */
   Future<bool> contains(Object needle) {
     _FutureImpl<bool> future = new _FutureImpl<bool>();
     StreamSubscription subscription;
     subscription = this.listen(
         (T element) {
           _runUserCode(
             () => (element == needle),
             (bool isMatch) {
               if (isMatch) {
-                subscription.cancel();
-                future._setValue(true);
+                subscription.cancel().whenComplete(() {
+                  future._setValue(true);
+                });
               }
             },
             _cancelAndError(subscription, future)
           );
         },
         onError: future._setError,
         onDone: () {
           future._setValue(false);
         },
         cancelOnError: true);
@@ skipping 34 matching lines @@
    */
   Future<bool> every(bool test(T element)) {
     _FutureImpl<bool> future = new _FutureImpl<bool>();
     StreamSubscription subscription;
     subscription = this.listen(
         (T element) {
           _runUserCode(
             () => test(element),
             (bool isMatch) {
               if (!isMatch) {
-                subscription.cancel();
-                future._setValue(false);
+                subscription.cancel().whenComplete(() {
+                  future._setValue(false);
+                });
               }
             },
             _cancelAndError(subscription, future)
           );
         },
         onError: future._setError,
         onDone: () {
           future._setValue(true);
         },
         cancelOnError: true);
     return future;
   }
 
   /**
    * Checks whether [test] accepts any element provided by this stream.
    *
    * Completes the [Future] when the answer is known.
    * If this stream reports an error, the [Future] will report that error.
    */
   Future<bool> any(bool test(T element)) {
     _FutureImpl<bool> future = new _FutureImpl<bool>();
     StreamSubscription subscription;
     subscription = this.listen(
         (T element) {
           _runUserCode(
             () => test(element),
             (bool isMatch) {
               if (isMatch) {
-                subscription.cancel();
-                future._setValue(true);
+                subscription.cancel().whenComplete(() {
+                  future._setValue(true);
+                });
               }
             },
             _cancelAndError(subscription, future)
           );
         },
         onError: future._setError,
         onDone: () {
           future._setValue(false);
         },
         cancelOnError: true);
@@ skipping 14 matching lines @@
       cancelOnError: true);
     return future;
   }
 
   /** Reports whether this stream contains any elements. */
   Future<bool> get isEmpty {
     _FutureImpl<bool> future = new _FutureImpl<bool>();
     StreamSubscription subscription;
     subscription = this.listen(
       (_) {
-        subscription.cancel();
-        future._setValue(false);
+        subscription.cancel().whenComplete(() {
+          future._setValue(false);
+        });
       },
       onError: future._setError,
       onDone: () {
         future._setValue(true);
       },
       cancelOnError: true);
     return future;
   }
 
   /** Collects the data of this stream in a [List]. */
@@ skipping 103 matching lines @@
    * Returns the first element.
    *
    * If [this] is empty throws a [StateError]. Otherwise this method is
    * equivalent to [:this.elementAt(0):]
    */
   Future<T> get first {
     _FutureImpl<T> future = new _FutureImpl<T>();
     StreamSubscription subscription;
     subscription = this.listen(
       (T value) {
-        subscription.cancel();
-        future._setValue(value);
-        return;
+        subscription.cancel().whenComplete(() {
+          future._setValue(value);
+        });
       },
       onError: future._setError,
       onDone: () {
         future._setError(new StateError("No elements"));
       },
       cancelOnError: true);
     return future;
   }
 
   /**
@@ skipping 29 matching lines @@
    * If [this] is empty or has more than one element throws a [StateError].
    */
   Future<T> get single {
     _FutureImpl<T> future = new _FutureImpl<T>();
     T result = null;
     bool foundResult = false;
     StreamSubscription subscription;
     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;
         result = value;
       },
       onError: future._setError,
       onDone: () {
         if (foundResult) {
           future._setValue(result);
           return;
@@ skipping 20 matching lines @@
    */
   Future<dynamic> firstWhere(bool test(T element), {Object defaultValue()}) {
     _FutureImpl<dynamic> future = new _FutureImpl();
     StreamSubscription subscription;
     subscription = this.listen(
       (T value) {
         _runUserCode(
           () => test(value),
           (bool isMatch) {
             if (isMatch) {
-              subscription.cancel();
-              future._setValue(value);
+              subscription.cancel().whenComplete(() {
+                future._setValue(value);
+              });
             }
           },
           _cancelAndError(subscription, future)
         );
       },
       onError: future._setError,
       onDone: () {
         if (defaultValue != null) {
           _runUserCode(defaultValue, future._setValue, future._setError);
           return;
@@ skipping 56 matching lines @@
     T result = null;
     bool foundResult = false;
     StreamSubscription subscription;
     subscription = this.listen(
       (T value) {
         _runUserCode(
           () => true == test(value),
           (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;
               result = value;
             }
           },
           _cancelAndError(subscription, future)
         );
       },
       onError: future._setError,
@@ skipping 16 matching lines @@
    * If this stream provides fewer than [index] elements before closing,
    * an error is reported.
    */
   Future<T> elementAt(int index) {
     if (index is! int || index < 0) throw new ArgumentError(index);
     _FutureImpl<T> future = new _FutureImpl<T>();
     StreamSubscription subscription;
     subscription = this.listen(
       (T value) {
         if (index == 0) {
-          subscription.cancel();
-          future._setValue(value);
+          subscription.cancel().whenComplete(() {
+            future._setValue(value);
+          });
           return;
         }
         index -= 1;
       },
       onError: future._setError,
       onDone: () {
         future._setError(new StateError("Not enough elements for elementAt"));
       },
       cancelOnError: true);
     return future;
   }
 }
 
 /**
  * A control object for the subscription on a [Stream].
  *
  * When you subscribe on a [Stream] using [Stream.listen],
  * a [StreamSubscription] object is returned. This object
  * is used to later unsubscribe again, or to temporarily pause
  * the stream's events.
  */
 abstract class StreamSubscription<T> {
   /**
    * Cancels this subscription. It will no longer receive events.
    *
    * 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));
 
   /** Set or override the error event handler of this subscription. */
   void onError(void handleError(error));
 
   /** Set or override the done event handler of this subscription. */
   void onDone(void handleDone());
 
@@ skipping 377 matching lines @@
    * The stream iterator is automatically canceled if the [moveNext] future
    * completes with either `false` or an error.
    *
    * If a [moveNext] call has been made, it will complete with `false` as value,
    * as will all further calls to [moveNext].
    *
    * If you need to stop listening for values before the stream iterator is
    * automatically closed, you must call [cancel] to ensure that the stream
    * is properly closed.
    */
-  void cancel();
+  Future cancel();
 }