Make await for cancel the stream when breaking out of the loop

runtime/lib/core_patch.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.
 
 import "dart:math";
 import "dart:typed_data";
 
 // Equivalent of calling FATAL from C++ code.
 _fatal(msg) native "DartCore_fatal";
 
@@ skipping 18 matching lines @@
 
 
 // _AsyncStarStreamController is used by the compiler to implement
 // async* generator functions.
 class _AsyncStarStreamController {
   StreamController controller;
   Function asyncStarBody;
   bool isAdding = false;
   bool onListenReceived = false;
   bool isScheduled = false;
+  Completer cancellationCompleter = null;
 
   Stream get stream => controller.stream;
 
   void runBody() {
     isScheduled = false;
     asyncStarBody();
   }
 
   void scheduleGenerator() {
     if (isScheduled || controller.isPaused || isAdding) {
@@ skipping 37 matching lines @@
     var whenDoneAdding =
         controller.addStream(stream as Stream, cancelOnError: false);
     whenDoneAdding.then((_) {
       isAdding = false;
       scheduleGenerator();
     });
     return false;
   }
 
   void addError(error, stackTrace) {
+    if ((cancellationCompleter != null &&

[regis, 2015/03/24 23:46:27]

Redundant pair of parenthesis

[hausner, 2015/03/24 23:57:29]

Done.

+        !cancellationCompleter.isCompleted)) {
+      // If the stream has been cancelled, complete the cancellation future
+      // with the error.
+      cancellationCompleter.completeError(error, stackTrace);
+      return;
+    }
     // If stream is cancelled, tell caller to exit the async generator.
     if (!controller.hasListener) return;
     controller.addError(error, stackTrace);
     // No need to schedule the generator body here. This code is only
     // called from the catch clause of the implicit try-catch-finally
     // around the generator body. That is, we are on the error path out
     // of the generator and do not need to run the generator again.
   }
 
   close() {
+    if ((cancellationCompleter != null) &&

[regis, 2015/03/24 23:46:27]

ditto

[hausner, 2015/03/24 23:57:29]

This one actually had the closing paren in the right place. Technically
redundant but I think our coding standards want a set of parens around the
(in)equality test.

+        !cancellationCompleter.isCompleted) {
+      // If the stream has been cancelled, complete the cancellation future
+      // with the error.
+      cancellationCompleter.complete();
+    }
     controller.close();
   }
 
   _AsyncStarStreamController(this.asyncStarBody) {
     controller = new StreamController(onListen: this.onListen,
                                       onResume: this.onResume,
                                       onCancel: this.onCancel);
   }
 
   onListen() {
     assert(!onListenReceived);
     onListenReceived = true;
     scheduleGenerator();
   }
 
   onResume() {
     scheduleGenerator();
   }
 
   onCancel() {
-    scheduleGenerator();
+    if (controller.isClosed) {
+      return null;
+    }
+    if (cancellationCompleter == null) {
+      cancellationCompleter = new Completer();
+      scheduleGenerator();
+    }
+    return cancellationCompleter.future;
   }
 }
 
 
 // _SyncIterable and _syncIterator are used by the compiler to
 // implement sync* generator functions. A sync* generator allocates
 // and returns a new _SyncIterable object.
 
 typedef bool SyncGeneratorCallback(Iterator iterator);
 
@@ skipping 37 matching lines @@
       if (isYieldEach) {
         // Spec mandates: it is a dynamic error if the class of [the object
         // returned by yield*] does not implement Iterable.
         yieldEachIterator = (current as Iterable).iterator;
         continue;
       }
       return true;
     }
   }
 }