Dart2js async* functions only resume execution of the body when suspended on yield.

sdk/lib/_internal/compiler/js_lib/js_helper.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.
 
 library _js_helper;
 
 import 'dart:_async_await_error_codes' as async_error_codes;
 
 import 'dart:_js_embedded_names' show
     DEFERRED_LIBRARY_URIS,
@@ skipping 3624 matching lines @@
   } else if (identical(bodyFunctionOrErrorCode, async_error_codes.ERROR)) {
     // The error is a js-error.
     completer.completeError(unwrapException(object),
                             getTraceFromException(object));
     return;
   }
   Future future = object is Future ? object : new Future.value(object);
   future.then(_wrapJsFunctionForAsync(bodyFunctionOrErrorCode,
                                       async_error_codes.SUCCESS),
       onError: (dynamic error, StackTrace stackTrace) {
-        ExceptionAndStackTrace wrapped =
+        ExceptionAndStackTrace wrappedException =
             new ExceptionAndStackTrace(error, stackTrace);
-        return _wrapJsFunctionForAsync(bodyFunctionOrErrorCode,
-                                       async_error_codes.ERROR)(wrapped);
+        Function wrapped =_wrapJsFunctionForAsync(bodyFunctionOrErrorCode,
+            async_error_codes.ERROR);
+        wrapped(wrappedException);
       });
   return completer.future;
 }
 
 Function _wrapJsFunctionForAsync(dynamic /* js function */ function,
                                  int errorCode) {
   var protected = JS('', """
     // Invokes [function] with [errorCode] and [result].
     //
     // If (and as long as) the invocation throws, calls [function] again,
@@ skipping 48 matching lines @@
 /// If the async* function wants to do an await it calls this function with
 /// [object] not and [IterationMarker].
 ///
 /// If [object] is not a [Future], it is wrapped in a `Future.value`.
 /// The [asyncBody] is called on completion of the future (see [asyncHelper].
 void asyncStarHelper(dynamic object,
                      dynamic /* int | js function */ bodyFunctionOrErrorCode,
                      AsyncStarStreamController controller) {
   if (identical(bodyFunctionOrErrorCode, async_error_codes.SUCCESS)) {
     // This happens on return from the async* function.
-    if (controller.cancelationCompleter != null) {
+    if (controller.isCanceled) {
       controller.cancelationCompleter.complete();
     } else {
       controller.close();
     }
     return;
   } else if (identical(bodyFunctionOrErrorCode, async_error_codes.ERROR)) {
     // The error is a js-error.
-    if (controller.cancelationCompleter != null) {
+    if (controller.isCanceled) {
       controller.cancelationCompleter.completeError(
           unwrapException(object),
           getTraceFromException(object));
     } else {
       controller.addError(unwrapException(object),
                           getTraceFromException(object));
       controller.close();
     }
     return;
   }
 
   if (object is IterationMarker) {
-    if (controller.cancelationCompleter != null) {
-      _wrapJsFunctionForAsync(bodyFunctionOrErrorCode,
-          async_error_codes.STREAM_WAS_CANCELED)(null);
+    if (controller.isCanceled) {
+      Function wrapped = _wrapJsFunctionForAsync(bodyFunctionOrErrorCode,
+          async_error_codes.STREAM_WAS_CANCELED);
+      wrapped(null);
       return;
     }
     if (object.state == IterationMarker.YIELD_SINGLE) {
       controller.add(object.value);
-      // If the controller is paused we stop producing more values.
-      if (controller.isPaused) {
-        return;
-      }
-      // TODO(sigurdm): We should not suspend here according to the spec.
+
       scheduleMicrotask(() {
-        _wrapJsFunctionForAsync(bodyFunctionOrErrorCode,
-                                async_error_codes.SUCCESS)
-            (null);
+        if (controller.isPaused) {
+          // We only suspend the thread inside the microtask in order to allow
+          // listeners on the output stream to pause in response to the just
+          // output value, and have the stream immediately stop producing.
+          controller.isSuspended = true;
+          return;
+        }
+        Function wrapped = _wrapJsFunctionForAsync(bodyFunctionOrErrorCode,
+            async_error_codes.SUCCESS);
+        wrapped(null);
       });
       return;
     } else if (object.state == IterationMarker.YIELD_STAR) {
       Stream stream = object.value;
-      controller.isAdding = true;
       // Errors of [stream] are passed though to the main stream. (see
-      // [AsyncStreamController.addStream].
+      // [AsyncStreamController.addStream]).
       // TODO(sigurdm): The spec is not very clear here. Clarify with Gilad.
       controller.addStream(stream).then((_) {
-        controller.isAdding = false;
-        _wrapJsFunctionForAsync(bodyFunctionOrErrorCode,
-                                async_error_codes.SUCCESS)(null);
+        // No check for isPaused here because the spec 17.16.2 only
+        // demands checks *before* each element in [stream] not after the last
+        // one. On the other hand we check for isCanceled, as that check happens
+        // after insertion of each element.
+        int errorCode = controller.isCanceled
+            ? async_error_codes.STREAM_WAS_CANCELED
+            : async_error_codes.SUCCESS;
+        Function wrapped = _wrapJsFunctionForAsync(bodyFunctionOrErrorCode,
+                                errorCode);
+        wrapped(null);
       });
       return;
     }
   }
 
   Future future = object is Future ? object : new Future.value(object);
   future.then(_wrapJsFunctionForAsync(bodyFunctionOrErrorCode,
                                       async_error_codes.SUCCESS),
               onError: (error, StackTrace stackTrace) {
-                ExceptionAndStackTrace wrapped =
+                ExceptionAndStackTrace wrappedException =
                     new ExceptionAndStackTrace(error, stackTrace);
-                return _wrapJsFunctionForAsync(bodyFunctionOrErrorCode,
-                                               async_error_codes.ERROR)
-                    (wrapped);
+                Function wrapped = _wrapJsFunctionForAsync(
+                    bodyFunctionOrErrorCode, async_error_codes.ERROR);
+                return wrapped(wrappedException);
               });
 }
 
 Stream streamOfController(AsyncStarStreamController controller) {
   return controller.stream;
 }
 
-/// A wrapper around a [StreamController] that remembers if that controller
-/// got a cancel.
+/// A wrapper around a [StreamController] that keeps track of the state of
+/// the execution of an async* function.
+/// It can be in 1 of 3 states:
 ///
-/// Also has a subSubscription that when not null will provide events for the
-/// stream, and will be paused and resumed along with this controller.
+/// - running/scheduled
+/// - suspended
+/// - canceled
+///
+/// If yielding while the subscription is paused it will become suspended. And
+/// only resume after the subscription is resumed or canceled.
 class AsyncStarStreamController {
   StreamController controller;
   Stream get stream => controller.stream;
+
+  /// True when the async* function has yielded while being paused.
+  /// When true execution will only resume after a `onResume` or `onCancel`
+  /// event.
+  bool isSuspended = false;
+
+  bool get isPaused => controller.isPaused;
+
   Completer cancelationCompleter = null;
+
+  /// True after the StreamSubscription has been cancelled.
+  /// When this is true, errors thrown from the async* body should go to the
+  /// [cancelationCompleter] instead of adding them to [controller], and
+  /// returning from the async function should complete [cancelationCompleter].
   bool get isCanceled => cancelationCompleter != null;
-  bool isAdding = false;
-  bool isPaused = false;
+
   add(event) => controller.add(event);
+
   addStream(Stream stream) {
     return controller.addStream(stream, cancelOnError: false);
   }
+
   addError(error, stackTrace) => controller.addError(error, stackTrace);
+
   close() => controller.close();
 
   AsyncStarStreamController(body) {
+
+    _resumeBody() {
+      scheduleMicrotask(() {
+        Function wrapped =
+            _wrapJsFunctionForAsync(body, async_error_codes.SUCCESS);
+        wrapped(null);
+      });
+    }
+
     controller = new StreamController(
       onListen: () {
-        scheduleMicrotask(() {
-          Function wrapped = _wrapJsFunctionForAsync(body,
-                                                     async_error_codes.SUCCESS);
-          wrapped(null);
-        });
-      },
-      onPause: () {
-        isPaused = true;
+        _resumeBody();
       }, onResume: () {
-        isPaused = false;
-        if (!isAdding) {
-          asyncStarHelper(null, body, this);
+        // Only schedule again if the async* function actually is suspended.
+        // Resume directly instead of scheduling, so that the sequence
+        // `pause-resume-pause` will result in one extra event produced.
+        if (isSuspended) {
+          isSuspended = false;
+          _resumeBody();
         }
       }, onCancel: () {
+        // If the async* is finished we ignore cancel events.
         if (!controller.isClosed) {
           cancelationCompleter = new Completer();
-          if (isPaused) asyncStarHelper(null, body, this);
-
+          if (isSuspended) {
+            // Resume the suspended async* function to run finalizers.
+            isSuspended = false;
+            scheduleMicrotask(() {
+              Function wrapped =_wrapJsFunctionForAsync(body,
+                  async_error_codes.STREAM_WAS_CANCELED);
+              wrapped(null);
+            });
+          }
           return cancelationCompleter.future;
         }
       });
   }
 }
 
 makeAsyncStarController(body) {
   return new AsyncStarStreamController(body);
 }
 
@@ skipping 95 matching lines @@
   // This is a function that will return a helper function that does the
   // iteration of the sync*.
   //
   // Each invocation should give a body with fresh state.
   final dynamic /* js function */ _outerHelper;
 
   SyncStarIterable(this._outerHelper);
 
   Iterator get iterator => new SyncStarIterator(JS('', '#()', _outerHelper));
 }