Reapply "Zone support for Futures, Timer, Streams."

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.expectCallback();
+  }
 
   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.cancelCallbackExpectation();
     _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.cancelCallbackExpectation();
+      _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 boundaries. It signals an
+ * uncaught error in the zone of origin when an error is sent from one error
+ * zone to another.
+ *
+ * When a Future is listening to another Future and they have not been
+ * instantiated in the same error-zone then Futures put an instance of this
+ * class between them (see [_FutureImpl._addListener]).
+ *
+ * For example:
+ *
+ *     var completer = new Completer();
+ *     var future = completer.future.then((x) => x);
+ *     catchErrors(() {
+ *       var future2 = future.catchError(print);
+ *     });
+ *     completer.completeError(499);
+ *
+ * In this example `future` and `future2` are in different error-zones. The
+ * error (499) that originates outside `catchErrors` must not reach the
+ * `catchError` future (`future2`) inside `catchErrors`.
+ *
+ * When invoking `catchError` on `future` the Future installs an
+ * [_ErrorZoneBoundaryListener] between itself and the result, `future2`.
+ *
+ * Conceptually _ErrorZoneBoundaryListeners could be implemented as
+ * `catchError`s on the origin future as well.
+ */
+class _ErrorZoneBoundaryListener implements _FutureListener {
+  _FutureListener _nextListener;
+  final _FutureListener _listener;
+
+  _ErrorZoneBoundaryListener(this._listener);
+
+  bool _inSameErrorZone(_Zone otherZone) {
+    // Should never be called. We use [_inSameErrorZone] to know if we have
+    // to insert an instance of [_ErrorZoneBoundaryListener] (and in the
+    // controller). Once we have inserted one we should never need to use it
+    // anymore.
+    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.handleUncaughtError(error);
+  }
 }
 
 class _FutureImpl<T> implements Future<T> {
   // State of the future. The state determines the interpretation of the
   // [resultOrListeners] field.
   // TODO(lrn): rename field since it can also contain a chained future.
 
   /// Initial state, waiting for a result. In this state, the
   /// [resultOrListeners] field holds a single-linked list of
   /// [FutureListener] listeners.
@@ skipping 15 matching lines @@
   /// Extra bit set when the future has been completed with an error result.
   /// but no listener has been scheduled to receive the error.
   /// If the bit is still set when a [runAsync] call triggers, the error will
   /// be reported to the top-level handler.
   /// Assigning a listener before that time will clear the bit.
   static const int _UNHANDLED_ERROR = 8;
 
   /** Whether the future is complete, and as what. */
   int _state = _INCOMPLETE;
 
+  final _Zone _zone = _Zone.current.fork();
+
   bool get _isChained => (_state & _CHAINED) != 0;
   bool get _hasChainedListener => _state == _CHAINED;
   bool get _isComplete => _state >= _VALUE;
   bool get _hasValue => _state == _VALUE;
   bool get _hasError => _state >= _ERROR;
   bool get _hasUnhandledError => _state >= _UNHANDLED_ERROR;
 
   void _clearUnhandledError() {
     _state &= ~_UNHANDLED_ERROR;
   }
@@ skipping 77 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 = _isChained ? null : _removeListeners();
     _state = _VALUE;
     _resultOrListeners = value;
     while (listeners != null) {
       _FutureListener listener = listeners;
       listeners = listener._nextListener;
       listener._nextListener = null;
       listener._sendValue(value);
@@ skipping 32 matching lines @@
     assert(_state == _ERROR);
     _state = _ERROR | _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();
         // TODO(floitsch): Hook this into unhandled error handling.
         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 (_isChained) {
       _state = _CHAINED;  // In case it was _CHAINED_UNLISTENED.
       _FutureImpl resultSource = _chainSource;
       resultSource._addListener(listener);
       return;
     }
     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 115 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.expectCallback();
+  }
 
-  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);
   }
 }