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 3696 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) {
+    if (controller.isCanceled) {
       _wrapJsFunctionForAsync(bodyFunctionOrErrorCode,
-          async_error_codes.STREAM_WAS_CANCELED)(null);
+                              async_error_codes.STREAM_WAS_CANCELED)
+          (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(() {
+        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;
+        }
         _wrapJsFunctionForAsync(bodyFunctionOrErrorCode,
                                 async_error_codes.SUCCESS)
             (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;
+        // We do not check for isPaused here because the spec 17.16.2 only
+        // demands checks *before* each element in [stream] not after the last
+        // one.

[Lasse Reichstein Nielsen, 2015/04/10 08:22:55]

Should you check for being cancelled?
The spec says to check for being canceled *after* adding each element to the
output stream (like isPaused is checked before).

The addStream may have stopped because the output stream was cancelled.
The spec does not expect that to happen automatically, rather it should happen
as the first part of the finally-handler that you skip to when seeing that the
output stream is cancelled, so you should probably check if the output is
cancelled and use .STREAM_WAS_CANCELED as argument instead of .SUCCESS if it is
- since you have already closed the input stream.

[sigurdm, 2015/04/10 08:58:19]

Done.

         _wrapJsFunctionForAsync(bodyFunctionOrErrorCode,
-                                async_error_codes.SUCCESS)(null);
+                                async_error_codes.SUCCESS)
+            (null);

[Lasse Reichstein Nielsen, 2015/04/10 08:22:55]

This way of emphasizing the call doesn't work for me, it's just confusing :)

How about either:
  (_wrapJsFunctionForAsync(bodyFunctionOrErrorCode,
                           async_error_codes.SUCCESS))(null);
or:
  var wrapped = _wrapJsFunctionForAsync(bodyFunctionOrErrorCode, 
                                        async_error_codes.SUCCESS);
  wrapped(null);

[sigurdm, 2015/04/10 08:58:19]

Done.

       });
       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 =
                     new ExceptionAndStackTrace(error, stackTrace);
                 return _wrapJsFunctionForAsync(bodyFunctionOrErrorCode,
                                                async_error_codes.ERROR)
                     (wrapped);
               });
 }
 
 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(() {
+              _wrapJsFunctionForAsync(body,
+                                      async_error_codes.STREAM_WAS_CANCELED)
+                  (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));
 }