Zone support for Futures.

sdk/lib/async/future_impl.dart

Index: sdk/lib/async/future_impl.dart
diff --git a/sdk/lib/async/future_impl.dart b/sdk/lib/async/future_impl.dart
index 3086f7d08afbea99d2dd271bfaadfda77ac011d6..70a82e7cbc3036088e6a40512d61cc6fd500014c 100644
--- a/sdk/lib/async/future_impl.dart
+++ b/sdk/lib/async/future_impl.dart
@@ -11,7 +11,10 @@ abstract class _Completer<T> implements Completer<T> {
   final Future<T> future;
   bool _isComplete = false;
 
-  _Completer() : future = new _FutureImpl<T>();
+  _Completer() : future = new _FutureImpl<T>() {
+    _FutureImpl futureImpl = future;
+    futureImpl._zone.incrementOpenCallbackCount();
+  }
 
   void _setFutureValue(T value);
   void _setFutureError(error);
@@ -19,6 +22,8 @@ abstract class _Completer<T> implements Completer<T> {
   void complete([T value]) {
     if (_isComplete) throw new StateError("Future already completed");
     _isComplete = true;
+    _FutureImpl futureImpl = future;
+    futureImpl._zone.decrementOpenCallbackCount();
     _setFutureValue(value);
   }
 
@@ -29,7 +34,13 @@ abstract class _Completer<T> implements Completer<T> {
       // Force the stack trace onto the error, even if it already had one.
       _attachStackTrace(error, stackTrace);
     }
-    _setFutureError(error);
+    _FutureImpl futureImpl = future;
+    if (futureImpl._inSameErrorZone(_Zone.current)) {
+      futureImpl._zone.decrementOpenCallbackCount();
+      _setFutureError(error);
+    } else {
+      _Zone.current.handleUncaughtError(error);
+    }
   }
 
   bool get isCompleted => _isComplete;
@@ -74,6 +85,8 @@ abstract class _FutureListener<T> {
   }
   void _sendValue(T value);
   void _sendError(error);
+
+  bool _inSameErrorZone(_Zone otherZone);
 }
 
 /** Adapter for a [_FutureImpl] to be a future result listener. */
@@ -83,6 +96,35 @@ class _FutureListenerWrapper<T> implements _FutureListener<T> {
   _FutureListenerWrapper(this.future);
   _sendValue(T value) { future._setValue(value); }
   _sendError(error) { future._setError(error); }
+  bool _inSameErrorZone(_Zone otherZone) => future._inSameErrorZone(otherZone);
+}
+
+/**
+ * This listener is installed at error-zone switches. It guarantees that no

[Lasse Reichstein Nielsen, 2013/05/21 08:19:57]

Need more explanation.
What is it installed on? By what? When is it called and by who?

[floitsch, 2013/05/21 18:08:32]

Done.

+ * error runs through zone boundaries that change error-handling.
+ */
+class _ErrorZoneBoundaryListener implements _FutureListener {
+  _FutureListener _nextListener;
+  final _FutureListener _listener;
+
+  _ErrorZoneBoundaryListener(this._listener) {
+    assert(!_nextListener._inSameErrorZone(_Zone.current));
+  }
+
+  bool _inSameErrorZone(_Zone otherZone) {

[Lasse Reichstein Nielsen, 2013/05/21 08:19:57]

Why is it not supported? Could it just return false?

[floitsch, 2013/05/21 18:08:32]

It's basically an assert. It could return anything, but in theory it should
never be called.
I added a comment, but I can also change it to return true or false.

+    throw new UnsupportedError(
+        "A Zone boundary doesn't support the inSameErrorZone test.");
+  }
+
+  void _sendValue(value) {
+    _listener._sendValue(value);
+  }
+
+  void _sendError(error) {
+    // We are not allowed to send an error from one error-zone to another.
+    // This is the whole purpose of this class.
+    _Zone.current._addError(error);
+  }
 }
 
 class _FutureImpl<T> implements Future<T> {
@@ -94,6 +136,8 @@ class _FutureImpl<T> implements Future<T> {
   /** Whether the future is complete, and as what. */
   int _state = _INCOMPLETE;
 
+  final _Zone _zone = _Zone.current.fork();
+
   bool get _isComplete => _state != _INCOMPLETE;
   bool get _hasValue => _state == _VALUE;
   bool get _hasError => (_state & _ERROR) != 0;
@@ -183,6 +227,10 @@ class _FutureImpl<T> implements Future<T> {
 
   Stream<T> asStream() => new Stream.fromFuture(this);
 
+  bool _inSameErrorZone(_Zone otherZone) {
+    return _zone.inSameErrorZone(otherZone);
+  }
+
   void _setValue(T value) {
     if (_isComplete) throw new StateError("Future already completed");
     _FutureListener listeners = _removeListeners();
@@ -222,17 +270,16 @@ class _FutureImpl<T> implements Future<T> {
         // No error handler has been added since the error was set.
         _clearUnhandledError();
         var error = _resultOrListeners;
-        print("Uncaught Error: ${error}");
-        var trace = getAttachedStackTrace(error);
-        if (trace != null) {
-          print("Stack Trace:\n$trace\n");
-        }
-        throw error;
+        _zone.handleUncaughtError(error);
       }
     });
   }
 
   void _addListener(_FutureListener listener) {
+    assert(listener._nextListener == null);
+    if (!listener._inSameErrorZone(_zone)) {
+      listener = new _ErrorZoneBoundaryListener(listener);
+    }
     if (_isComplete) {
       _clearUnhandledError();
       // Handle late listeners asynchronously.
@@ -247,7 +294,6 @@ class _FutureImpl<T> implements Future<T> {
       });
     } else {
       assert(!_isComplete);
-      assert(listener._nextListener == null);
       listener._nextListener = _resultOrListeners;
       _resultOrListeners = listener;
     }
@@ -328,13 +374,24 @@ abstract class _TransformFuture<S, T> extends _FutureImpl<T>
   // _FutureListener implementation.
   _FutureListener _nextListener;
 
-  void _sendValue(S value);
+  _TransformFuture() {
+    _zone.incrementOpenCallbackCount();
+  }
 
-  void _sendError(error);
+  void _sendValue(S value) {
+    _zone.executeCallback(() => _zonedSendValue(value));
+  }
+
+  void _sendError(error) {
+    _zone.executeCallback(() => _zonedSendError(error));
+  }
 
   void _subscribeTo(_FutureImpl future) {
     future._addListener(this);
   }
+
+  void _zonedSendValue(S value);
+  void _zonedSendError(error);
 }
 
 /** The onValue and onError handlers return either a value or a future */
@@ -353,7 +410,7 @@ class _ThenFuture<S, T> extends _TransformFuture<S, T> {
 
   _ThenFuture(this._onValue);
 
-  _sendValue(S value) {
+  _zonedSendValue(S value) {
     assert(_onValue != null);
     var result;
     try {
@@ -365,7 +422,7 @@ class _ThenFuture<S, T> extends _TransformFuture<S, T> {
     _setOrChainValue(result);
   }
 
-  void _sendError(error) {
+  void _zonedSendError(error) {
     _setError(error);
   }
 }
@@ -377,11 +434,11 @@ class _CatchErrorFuture<T> extends _TransformFuture<T,T> {
 
   _CatchErrorFuture(this._onError, this._test);
 
-  _sendValue(T value) {
+  _zonedSendValue(T value) {
     _setValue(value);
   }
 
-  _sendError(error) {
+  _zonedSendError(error) {
     assert(_onError != null);
     // if _test is supplied, check if it returns true, otherwise just
     // forward the error unmodified.
@@ -418,7 +475,7 @@ class _SubscribeFuture<S, T> extends _ThenFuture<S, T> {
 
   // The _sendValue method is inherited from ThenFuture.
 
-  void _sendError(error) {
+  void _zonedSendError(error) {
     assert(_onError != null);
     var result;
     try {
@@ -437,7 +494,7 @@ class _WhenFuture<T> extends _TransformFuture<T, T> {
 
   _WhenFuture(this._action);
 
-  void _sendValue(T value) {
+  void _zonedSendValue(T value) {
     try {
       var result = _action();
       if (result is Future) {
@@ -454,7 +511,7 @@ class _WhenFuture<T> extends _TransformFuture<T, T> {
     _setValue(value);
   }
 
-  void _sendError(error) {
+  void _zonedSendError(error) {
     try {
       var result = _action();
       if (result is Future) {