feat(scope): Use VmTurnZone.onScheduleMicrotask in Scope

and

fix(scope): Use runAsync for microtasks in digest.

BREAKING CHANGE:

Microtasks scheduled in flush will process in current cycle, but they
are not allowed to do model changes.

Microtasks scheduled in digest will be executed in digest, counting
towards the ScopeDigestTTL.

Author: mvuksano

lib/core/scope.dart

@@ -65,7 +65,7 @@ class ScopeEvent {
 @Injectable()
 class ScopeDigestTTL {
   final int ttl;
-  ScopeDigestTTL(): ttl = 5;
+  ScopeDigestTTL(): ttl = 10;
   ScopeDigestTTL.value(this.ttl);
 }
 
@@ -657,6 +657,7 @@ class RootScope extends Scope {
   {
     _zone.onTurnDone = apply;
     _zone.onError = (e, s, ls) => _exceptionHandler(e, s);
+    _zone.onScheduleMicrotask = runAsync;
     cacheRegister.registerCache("ScopeWatchASTs", astCache);
   }
 
@@ -693,15 +694,8 @@ class RootScope extends Scope {
       ChangeLog changeLog;
       _scopeStats.digestStart();
       do {
-        while (_runAsyncHead != null) {
-          try {
-            _runAsyncHead.fn();
-          } catch (e, s) {
-            _exceptionHandler(e, s);
-          }
-          _runAsyncHead = _runAsyncHead._next;
-        }
-        _runAsyncTail = null;
+
+        int asyncCount = _runAsyncFns();
 
         digestTTL--;
         count = rootWatchGroup.detectChanges(
@@ -717,7 +711,7 @@ class RootScope extends Scope {
             digestLog = [];
             changeLog = (e, c, p) => digestLog.add('$e: $c <= $p');
           } else {
-            log.add(digestLog.join(', '));
+            log.add("${asyncCount > 0 ? 'async:$asyncCount' : ''}${digestLog.join(', ')}");
             digestLog.clear();
           }
         }
@@ -726,7 +720,7 @@ class RootScope extends Scope {
                 'Last $LOG_COUNT iterations:\n${log.join('\n')}';
         }
         _scopeStats.digestLoop(count);
-      } while (count > 0);
+      } while (count > 0 || _runAsyncHead != null);
     } finally {
       _scopeStats.digestEnd();
       _transitionState(STATE_DIGEST, null);
@@ -769,7 +763,8 @@ class RootScope extends Scope {
           if (_domReadHead == null) _stats.domReadEnd();
         }
         _domReadTail = null;
-      } while (_domWriteHead != null || _domReadHead != null);
+        _runAsyncFns();
+      } while (_domWriteHead != null || _domReadHead != null || _runAsyncHead != null);
       _stats.flushEnd();
       assert((() {
         _stats.flushAssertStart();
@@ -801,6 +796,9 @@ class RootScope extends Scope {
 
   // QUEUES
   void runAsync(fn()) {
+    if (_state == STATE_FLUSH_ASSERT) {
+      throw "Scheduling microtasks not allowed in $state state.";
+    }
     var chain = new _FunctionChain(fn);
     if (_runAsyncHead == null) {
       _runAsyncHead = _runAsyncTail = chain;
@@ -809,6 +807,21 @@ class RootScope extends Scope {
     }
   }
 
+  _runAsyncFns() {
+    var count = 0;
+    while (_runAsyncHead != null) {
+      try {
+        count++;
+        _runAsyncHead.fn();
+      } catch (e, s) {
+        _exceptionHandler(e, s);
+      }
+      _runAsyncHead = _runAsyncHead._next;
+    }
+    _runAsyncTail = null;
+    return count;
+  }
+
   void domWrite(fn()) {
     var chain = new _FunctionChain(fn);
     if (_domWriteHead == null) {

lib/core/zone.dart

@@ -60,6 +60,7 @@ class VmTurnZone {
   /// an "inner" [Zone], which is a child of the outer [Zone].
   async.Zone _innerZone;
 
+
   /**
    * Associates with this
    *

test/core/scope_spec.dart

@@ -1226,6 +1226,20 @@ void main() {
       });
 
 
+      it(r'should detect infinite digest through runAsync', (RootScope rootScope) {
+        rootScope.context['value'] = () { rootScope.runAsync(() {}); return 'a'; };
+        rootScope.watch('value()', (_, __) {});
+
+        expect(() {
+          rootScope.digest();
+        }).toThrow('Model did not stabilize in 5 digests. '
+        'Last 3 iterations:\n'
+        'async:1\n'
+        'async:1\n'
+        'async:1');
+      });
+
+
       it(r'should always call the watchr with newVal and oldVal equal on the first run',
           (RootScope rootScope) {
         var log = [];
@@ -1439,6 +1453,105 @@ void main() {
       });
     });
 
+    describe('microtask processing', () {
+      beforeEach((VmTurnZone zone, RootScope scope, Logger log) {
+        var onTurnDone = zone.onTurnDone;
+        zone.onTurnDone = () {
+          log('[');
+          onTurnDone();
+          log(']');
+        };
+        var onScheduleMicrotask = zone.onScheduleMicrotask;
+        zone.onScheduleMicrotask = (fn) {
+          log('(');
+          try {
+            onScheduleMicrotask(fn);
+          } catch (e) {
+            log('CATCH: $e');
+          }
+          log(')');
+        };
+      });
+
+      it('should schedule apply after future resolution',
+        async((Logger log, VmTurnZone zone, RootScope scope) {
+          Completer completer;
+          zone.run(() {
+            completer = new Completer();
+            completer.future.then((value) {
+              log('then($value)');
+            });
+          });
+
+          scope.runAsync(() => log('before'));
+          log.clear();
+          completer.complete('OK'); // this one causes APPLY which processe 'before'
+          // This one schedules work but apply already run so it does not execute.
+          scope.runAsync(() => log('NOT_EXECUTED'));
+
+          expect(log).toEqual(['(', ')', '[', 'before', 'then(OK)', ']']);
+        })
+      );
+
+      it('should schedule microtask to runAsync queue during digest',
+        async((Logger log, VmTurnZone zone, RootScope scope) {
+          Completer completer;
+          zone.run(() {
+            completer = new Completer();
+            completer.future.
+              then((value) {
+                scope.runAsync(() => log('in(${scope.state})'));
+                return new Future.value(value);
+              }).
+              then((value) {
+                log('then($value)');
+              });
+          });
+          log.clear();
+          completer.complete('OK');
+          expect(log).toEqual(['(', ')', '[', '(', ')', 'in(digest)', 'then(OK)', ']']);
+        })
+      );
+
+      it('should allow microtasks in flush phase and process them immediatly',
+        async((Logger log, VmTurnZone zone, RootScope scope) {
+          scope.watch('g()', (_, __) {});
+          scope.context['g'] = () {
+            log('!');
+            return 0;
+          };
+
+          zone.run(() {
+            scope.domWrite(() {
+              log('domWriteA');
+              return new Future.value(null).then((_) => scope.domWrite(() => log('domWriteB')));
+            });
+          });
+          expect(log).toEqual(
+              ['[', '!', '!', 'domWriteA', '(', ')', 'domWriteB', /* assert */'!', ']']);
+        })
+      );
+
+      it('should allow creation of Completers in flush phase',
+        async((Logger log, VmTurnZone zone, RootScope scope) {
+          Completer completer;
+          zone.run(() {
+            scope.domWrite(() {
+              log('new Completer');
+              completer = new Completer();
+              completer.future.then((value) {
+                log('then($value)');
+              });
+            });
+          });
+          log('=');
+          completer.complete('OK');
+          log(';');
+          expect(log).toEqual(
+              ['[', 'new Completer', ']', '=', '(', ')', '[', 'then(OK)', ']', ';']);
+        })
+      );
+    });
 
     describe('domRead/domWrite', () {
       beforeEachModule((Module module) {