Add error-intercept for Completer.completeError and StreamController.addError.

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;
 
 /** The onValue and onError handlers return either a value or a future */
 typedef dynamic _FutureOnValue<T>(T value);
 /** Test used by [Future.catchError] to handle skip some errors. */
 typedef bool _FutureErrorTest(var error);
 /** Used by [WhenFuture]. */
 typedef _FutureAction();
 
 abstract class _Completer<T> implements Completer<T> {
   final _Future<T> future = new _Future<T>();
 
   void complete([value]);
 
-  void completeError(Object error, [StackTrace stackTrace]);
+  void completeError(Object error, [StackTrace stackTrace]) {
+    if (error == null) throw new ArgumentError("Error must not be null");
+    if (!future._mayComplete) throw new StateError("Future already completed");
+    AsyncError replacement = Zone.current.errorCallback(error, stackTrace);
+    if (replacement != null) {
+      error = replacement.error;
+      stackTrace = replacement.stackTrace;
+    }
+    _completeError(error, stackTrace);
+  }
+
+  void _completeError(Object error, StackTrace stackTrace);
 
   // The future's _isComplete doesn't take into account pending completions.
   // We therefore use _mayComplete.
   bool get isCompleted => !future._mayComplete;
 }
 
 class _AsyncCompleter<T> extends _Completer<T> {
 
   void complete([value]) {
     if (!future._mayComplete) throw new StateError("Future already completed");
     future._asyncComplete(value);
   }
 
-  void completeError(Object error, [StackTrace stackTrace]) {
-    if (error == null) throw new ArgumentError("Error must not be null");
-    if (!future._mayComplete) throw new StateError("Future already completed");
+  void _completeError(Object error, StackTrace stackTrace) {
     future._asyncCompleteError(error, stackTrace);
   }
 }
 
 class _SyncCompleter<T> extends _Completer<T> {
 
   void complete([value]) {
     if (!future._mayComplete) throw new StateError("Future already completed");
     future._complete(value);
   }
 
-  void completeError(Object error, [StackTrace stackTrace]) {
-    if (!future._mayComplete) throw new StateError("Future already completed");
+  void _completeError(Object error, StackTrace stackTrace) {
     future._completeError(error, stackTrace);
   }
 }
 
 class _Future<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
@@ skipping 147 matching lines @@
   void _markPendingCompletion() {
     if (!_mayComplete) throw new StateError("Future already completed");
     _state = _PENDING_COMPLETE;
   }
 
   T get _value {
     assert(_isComplete && _hasValue);
     return _resultOrListeners;
   }
 
-  _AsyncError get _error {
+  AsyncError get _error {
     assert(_isComplete && _hasError);
     return _resultOrListeners;
   }
 
   void _setValue(T value) {
     assert(!_isComplete);  // But may have a completion pending.
     _state = _VALUE;
     _resultOrListeners = value;
   }
 
   void _setError(Object error, StackTrace stackTrace) {
     assert(!_isComplete);  // But may have a completion pending.
     _state = _ERROR;
-    _resultOrListeners = new _AsyncError(error, stackTrace);
+    _resultOrListeners = new AsyncError(error, stackTrace);
   }
 
   void _addListener(_Future listener) {
     assert(listener._nextListener == null);
     if (_isComplete) {
       // Handle late listeners asynchronously.
       _zone.scheduleMicrotask(() {
         _propagateToListeners(this, listener);
       });
     } else {
@@ skipping 189 matching lines @@
    *
    * If [runCallback] is true (which should be the default) it executes
    * the registered action of listeners. If it is `false` then the callback is
    * skipped. This is used to complete futures with chained futures.
    */
   static void _propagateToListeners(_Future source, _Future listeners) {
     while (true) {
       if (!source._isComplete) return;  // Chained future.
       bool hasError = source._hasError;
       if (hasError && listeners == null) {
-        _AsyncError asyncError = source._error;
+        AsyncError asyncError = source._error;
         source._zone.handleUncaughtError(
             asyncError.error, asyncError.stackTrace);
         return;
       }
       if (listeners == null) return;
       _Future listener = listeners;
       if (listener._nextListener != null) {
         // Usually futures only have one listener. If they have several, we
         // handle them specially.
         _propagateMultipleListeners(source, listeners);
@@ skipping 13 matching lines @@
       // the future is not already marked (or chained).
       // Only if we either have an error or callbacks, go into this, somewhat
       // expensive, branch. Here we'll enter/leave the zone. Many futures
       // doesn't have callbacks, so this is a significant optimization.
       if (hasError ||
           listener._onValue != null ||
           listener._whenCompleteAction != null) {
         Zone zone = listener._zone;
         if (hasError && !source._zone.inSameErrorZone(zone)) {
           // Don't cross zone boundaries with errors.
-          _AsyncError asyncError = source._error;
+          AsyncError asyncError = source._error;
           source._zone.handleUncaughtError(
               asyncError.error, asyncError.stackTrace);
           return;
         }
 
         Zone oldZone;
         if (!identical(Zone.current, zone)) {
           // Change zone if it's not current.
           oldZone = Zone._enter(zone);
         }
 
         bool handleValueCallback() {
           try {
             listenerValueOrError = zone.runUnary(listener._onValue,
                                                  sourceValue);
             return true;
           } catch (e, s) {
-            listenerValueOrError = new _AsyncError(e, s);
+            listenerValueOrError = new AsyncError(e, s);
             return false;
           }
         }
 
         void handleError() {
-          _AsyncError asyncError = source._error;
+          AsyncError asyncError = source._error;
           _FutureErrorTest test = listener._errorTest;
           bool matchesTest = true;
           if (test != null) {
             try {
               matchesTest = zone.runUnary(test, asyncError.error);
             } catch (e, s) {
               // TODO(ajohnsen): Should we suport rethrow for test throws?
               listenerValueOrError = identical(asyncError.error, e) ?
-                  asyncError : new _AsyncError(e, s);
+                  asyncError : new AsyncError(e, s);
               listenerHasValue = false;
               return;
             }
           }
           Function errorCallback = listener._onError;
           if (matchesTest && errorCallback != null) {
             try {
               if (errorCallback is ZoneBinaryCallback) {
                 listenerValueOrError = zone.runBinary(errorCallback,
                                                       asyncError.error,
                                                       asyncError.stackTrace);
               } else {
                 listenerValueOrError = zone.runUnary(errorCallback,
                                                      asyncError.error);
               }
             } catch (e, s) {
               listenerValueOrError = identical(asyncError.error, e) ?
-                  asyncError : new _AsyncError(e, s);
+                  asyncError : new AsyncError(e, s);
               listenerHasValue = false;
               return;
             }
             listenerHasValue = true;
           } else {
             // Copy over the error from the source.
             listenerValueOrError = asyncError;
             listenerHasValue = false;
           }
         }
 
         void handleWhenCompleteCallback() {
           var completeResult;
           try {
             completeResult = zone.run(listener._whenCompleteAction);
           } catch (e, s) {
             if (hasError && identical(source._error.error, e)) {
               listenerValueOrError = source._error;
             } else {
-              listenerValueOrError = new _AsyncError(e, s);
+              listenerValueOrError = new AsyncError(e, s);
             }
             listenerHasValue = false;
           }
           if (completeResult is Future) {
             listener._isChained = true;
             isPropagationAborted = true;
             completeResult.then((ignored) {
               // Try again. Since the future is marked as chained it won't run
               // the whenComplete again.
               _propagateToListeners(source, listener);
@@ skipping 51 matching lines @@
             _chainForeignFuture(chainSource, listener);
           }
           return;
         }
       }
       if (listenerHasValue) {
         listeners = listener._removeListeners();
         listener._setValue(listenerValueOrError);
       } else {
         listeners = listener._removeListeners();
-        _AsyncError asyncError = listenerValueOrError;
+        AsyncError asyncError = listenerValueOrError;
         listener._setError(asyncError.error, asyncError.stackTrace);
       }
       // Prepare for next round.
       source = listener;
     }
   }
 
   Future timeout(Duration timeLimit, {onTimeout()}) {
     if (_isComplete) return new _Future.immediate(this);
     _Future result = new _Future();
@@ skipping 21 matching lines @@
       }
     }, onError: (e, s) {
       if (timer.isActive) {
         timer.cancel();
         result._completeError(e, s);
       }
     });
     return result;
   }
 }