Implement async* functions in VM

runtime/lib/core_patch.dart

Index: runtime/lib/core_patch.dart
===================================================================
--- runtime/lib/core_patch.dart	(revision 43883)
+++ runtime/lib/core_patch.dart	(working copy)
@@ -22,11 +22,77 @@
   String toString() => _enum_names[index];
 }
 
-typedef bool SyncGeneratorCallback(Iterator iterator);
 
+// _AsyncStarStreamController is used by the compiler to implement
+// async* generator functions.
+class _AsyncStarStreamController {
+  StreamController controller;
+  Function asyncStarBody;
+  bool isCancelled = false;
+  bool isAdding = false;
+
+  Stream get stream => controller.stream;
+  //bool get isPaused => isAdding || controller.isPaused;
+
+  // Adds element to steam, returns true iff the caller should suspend
+  // execution of the generator.
+  bool add(event) {
+    controller.add(event);
+    return controller.isPaused;
+  }
+
+  // Adds stream, tells caller to suspend execution of the generator
+  // function. The generator will be scheduled again when all of the
+  // elements of the added stream have been consumed.
+  bool addStream(Stream stream) {
+    isAdding = true;
+    var whenDoneAdding =
+        controller.addStream(stream as Stream, cancelOnError: false);

[Ivan Posva, 2015/02/23 08:23:36]

Why is there an "as Stream" here?

[hausner, 2015/02/23 17:12:01]

Because the spec says it is a runtime error if yield* returns a value that is
not a stream. The type annotation on the parameter does not fire in non-checked
mode. The as-operator does a type check even in non-checked mode.

+    whenDoneAdding.then((_) {
+      isAdding = false;
+      if (!controller.isPaused) {
+        scheduleMicrotask(asyncStarBody);
+      }
+    });
+    return true;
+  }
+
+  addError(error, stackTrace) {
+    controller.addError(error, stackTrace);
+  }
+
+  close() {
+    controller.close();
+  }
+
+  _AsyncStarStreamController(this.asyncStarBody) {
+    controller = new StreamController(onListen: this.onListen,
+                                      onResume: this.onResume,
+                                      onCancel: this.onCancel);
+  }
+
+  onListen() {
+    scheduleMicrotask(asyncStarBody);
+  }
+
+  onResume() {
+    if (!isAdding) {
+      scheduleMicrotask(asyncStarBody);
+    }
+  }
+
+  onCancel() {
+    isCancelled = true;
+  }
+}
+
+
 // _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);
+
 class _SyncIterable extends IterableBase {
   // moveNextFn is the closurized body of the generator function.
   final SyncGeneratorCallback moveNextFn;