Zone support for Futures.

sdk/lib/async/future_impl.dart

 // Copyright (c) 2012, 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.
 
 part of dart.async;
 
 deprecatedFutureValue(_FutureImpl future) =>
   future._isComplete ? future._resultOrListeners : null;
 
 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);
 
   void complete([T value]) {
     if (_isComplete) throw new StateError("Future already completed");
     _isComplete = true;
+    _FutureImpl futureImpl = future;
+    futureImpl._zone.decrementOpenCallbackCount();
     _setFutureValue(value);
   }
 
   void completeError(Object error, [Object stackTrace = null]) {
     if (_isComplete) throw new StateError("Future already completed");
     _isComplete = true;
     if (stackTrace != null) {
       // 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;
 }
 
 class _AsyncCompleter<T> extends _Completer<T> {
   void _setFutureValue(T value) {
     _FutureImpl future = this.future;
     runAsync(() { future._setValue(value); });
   }
@@ skipping 24 matching lines @@
  *
  * Listeners are kept in a linked list.
  */
 abstract class _FutureListener<T> {
   _FutureListener _nextListener;
   factory _FutureListener.wrap(_FutureImpl future) {
     return new _FutureListenerWrapper(future);
   }
   void _sendValue(T value);
   void _sendError(error);
+
+  bool _inSameErrorZone(_Zone otherZone);
 }
 
 /** Adapter for a [_FutureImpl] to be a future result listener. */
 class _FutureListenerWrapper<T> implements _FutureListener<T> {
   _FutureImpl future;
   _FutureListener _nextListener;
   _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> {
   static const int _INCOMPLETE = 0;
   static const int _VALUE = 1;
   static const int _ERROR = 2;
   static const int _UNHANDLED_ERROR = 4;
 
   /** 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;
   bool get _hasUnhandledError => (_state & _UNHANDLED_ERROR) != 0;
 
   void _clearUnhandledError() {
     // Works because _UNHANDLED_ERROR is highest bit in use.
     _state &= ~_UNHANDLED_ERROR;
   }
 
@@ skipping 69 matching lines @@
   Future catchError(f(error), { bool test(error) }) {
     return new _CatchErrorFuture(f, test).._subscribeTo(this);
   }
 
   Future<T> whenComplete(action()) {
     return new _WhenFuture<T>(action).._subscribeTo(this);
   }
 
   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();
     _state = _VALUE;
     _resultOrListeners = value;
     while (listeners != null) {
       _FutureListener listener = listeners;
       listeners = listener._nextListener;
       listener._nextListener = null;
       listener._sendValue(value);
@@ skipping 19 matching lines @@
 
   void _scheduleUnhandledError() {
     _state |= _UNHANDLED_ERROR;
     // Wait for the rest of the current event's duration to see
     // if a subscriber is added to handle the error.
     runAsync(() {
       if (_hasUnhandledError) {
         // 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.
       runAsync(() {
         if (_hasValue) {
           T value = _resultOrListeners;
           listener._sendValue(value);
         } else {
           assert(_hasError);
           listener._sendError(_resultOrListeners);
         }
       });
     } else {
       assert(!_isComplete);
-      assert(listener._nextListener == null);
       listener._nextListener = _resultOrListeners;
       _resultOrListeners = listener;
     }
   }
 
   _FutureListener _removeListeners() {
     // Reverse listeners before returning them, so the resulting list is in
     // subscription order.
     assert(!_isComplete);
     _FutureListener current = _resultOrListeners;
@@ skipping 60 matching lines @@
  *
  * A transforming future is itself a future and a future listener.
  * Subclasses override [_sendValue]/[_sendError] to intercept
  * the results of a previous future.
  */
 abstract class _TransformFuture<S, T> extends _FutureImpl<T>
                                       implements _FutureListener<S> {
   // _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 */
 typedef dynamic _FutureOnValue<T>(T value);
 typedef dynamic _FutureOnError(error);
 /** Test used by [Future.catchError] to handle skip some errors. */
 typedef bool _FutureErrorTest(var error);
 /** Used by [WhenFuture]. */
 typedef _FutureAction();
 
 /** Future returned by [Future.then] with no [:onError:] parameter. */
 class _ThenFuture<S, T> extends _TransformFuture<S, T> {
   // TODO(ahe): Restore type when feature is implemented in dart2js
   // checked mode.
   final /* _FutureOnValue<S> */ _onValue;
 
   _ThenFuture(this._onValue);
 
-  _sendValue(S value) {
+  _zonedSendValue(S value) {
     assert(_onValue != null);
     var result;
     try {
       result = _onValue(value);
     } catch (e, s) {
       _setError(_asyncError(e, s));
       return;
     }
     _setOrChainValue(result);
   }
 
-  void _sendError(error) {
+  void _zonedSendError(error) {
     _setError(error);
   }
 }
 
 /** Future returned by [Future.catchError]. */
 class _CatchErrorFuture<T> extends _TransformFuture<T,T> {
   final _FutureErrorTest _test;
   final _FutureOnError _onError;
 
   _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.
     if (_test != null) {
       bool matchesTest;
       try {
         matchesTest = _test(error);
       } catch (e, s) {
         _setError(_asyncError(e, s));
         return;
@@ skipping 16 matching lines @@
 }
 
 /** Future returned by [Future.then] with an [:onError:] parameter. */
 class _SubscribeFuture<S, T> extends _ThenFuture<S, T> {
   final _FutureOnError _onError;
 
   _SubscribeFuture(onValue(S value), this._onError) : super(onValue);
 
   // The _sendValue method is inherited from ThenFuture.
 
-  void _sendError(error) {
+  void _zonedSendError(error) {
     assert(_onError != null);
     var result;
     try {
       result = _onError(error);
     } catch (e, s) {
       _setError(_asyncError(e, s));
       return;
     }
     _setOrChainValue(result);
   }
 }
 
 /** Future returned by [Future.whenComplete]. */
 class _WhenFuture<T> extends _TransformFuture<T, T> {
   final _FutureAction _action;
 
   _WhenFuture(this._action);
 
-  void _sendValue(T value) {
+  void _zonedSendValue(T value) {
     try {
       var result = _action();
       if (result is Future) {
         Future resultFuture = result;
         resultFuture.then((_) {
           _setValue(value);
         }, onError: _setError);
         return;
       }
     } catch (e, s) {
       _setError(_asyncError(e, s));
       return;
     }
     _setValue(value);
   }
 
-  void _sendError(error) {
+  void _zonedSendError(error) {
     try {
       var result = _action();
       if (result is Future) {
         Future resultFuture = result;
         // TODO(lrn): Find a way to combine [error] into [e].
         resultFuture.then((_) {
           _setError(error);
         }, onError: _setError);
         return;
       }
     } catch (e, s) {
       error = _asyncError(e, s);
     }
     _setError(error);
   }
 }