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

sdk/lib/async/stream_pipe.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;
 
 /** Runs user code and takes actions depending on success or failure. */
 _runUserCode(userCode(),
              onSuccess(value),
              onError(error, StackTrace stackTrace)) {
   try {
     onSuccess(userCode());
   } catch (e, s) {
-    onError(e, s);
+    AsyncError replacement = Zone.current.errorCallback(e, s);
+    if (replacement == null) {
+      onError(e, s);
+    } else {
+      onError(replacement.error, replacement.stackTrace);
+    }
   }
 }
 
 /** Helper function to cancel a subscription and wait for the potential future,
   before completing with an error. */
 void _cancelAndError(StreamSubscription subscription,
                      _Future future,
                      error,
                      StackTrace stackTrace) {
   var cancelFuture = subscription.cancel();
   if (cancelFuture is Future) {
     cancelFuture.whenComplete(() => future._completeError(error, stackTrace));
   } else {
     future._completeError(error, stackTrace);
   }
 }
 
+void _cancelAndErrorWithReplacement(StreamSubscription subscription,
+                                    _Future future,
+                                    error, StackTrace stackTrace) {
+  AsyncError replacement = Zone.current.errorCallback(error, stackTrace);
+  if (replacement != null) {
+    error = replacement.error;
+    stackTrace = replacement.stackTrace;
+  }
+  _cancelAndError(subscription, future, error, stackTrace);
+}
+
 /** Helper function to make an onError argument to [_runUserCode]. */
 _cancelAndErrorClosure(StreamSubscription subscription, _Future future) =>
   ((error, StackTrace stackTrace) => _cancelAndError(
       subscription, future, error, stackTrace));
 
 /** Helper function to cancel a subscription and wait for the potential future,
   before completing with a value. */
 void _cancelAndValue(StreamSubscription subscription, _Future future, value) {
   var cancelFuture = subscription.cancel();
   if (cancelFuture is Future) {
@@ skipping 120 matching lines @@
     _stream._handleDone(this);
   }
 }
 
 // -------------------------------------------------------------------
 // Stream transformers used by the default Stream implementation.
 // -------------------------------------------------------------------
 
 typedef bool _Predicate<T>(T value);
 
+void _addErrorWithReplacement(_EventSink sink, error, stackTrace) {
+  AsyncError replacement = Zone.current.errorCallback(error, stackTrace);
+  if (replacement == null) {
+    sink._addError(error, stackTrace);
+  } else {
+    sink._addError(replacement.error, replacement.stackTrace);
+  }
+}
+
+
 class _WhereStream<T> extends _ForwardingStream<T, T> {
   final _Predicate<T> _test;
 
   _WhereStream(Stream<T> source, bool test(T value))
       : _test = test, super(source);
 
   void _handleData(T inputEvent, _EventSink<T> sink) {
     bool satisfies;
     try {
       satisfies = _test(inputEvent);
     } catch (e, s) {
-      sink._addError(e, s);
+      _addErrorWithReplacement(sink, e, s);
       return;
     }
     if (satisfies) {
       sink._add(inputEvent);
     }
   }
 }
 
 
 typedef T _Transformation<S, T>(S value);
 
 /**
  * A stream pipe that converts data events before passing them on.
  */
 class _MapStream<S, T> extends _ForwardingStream<S, T> {
   final _Transformation _transform;
 
   _MapStream(Stream<S> source, T transform(S event))
       : this._transform = transform, super(source);
 
   void _handleData(S inputEvent, _EventSink<T> sink) {
     T outputEvent;
     try {
       outputEvent = _transform(inputEvent);
     } catch (e, s) {
-      sink._addError(e, s);
+      _addErrorWithReplacement(sink, e, s);
       return;
     }
     sink._add(outputEvent);
   }
 }
 
 /**
  * A stream pipe that converts data events before passing them on.
  */
 class _ExpandStream<S, T> extends _ForwardingStream<S, T> {
   final _Transformation<S, Iterable<T>> _expand;
 
   _ExpandStream(Stream<S> source, Iterable<T> expand(S event))
       : this._expand = expand, super(source);
 
   void _handleData(S inputEvent, _EventSink<T> sink) {
     try {
       for (T value in _expand(inputEvent)) {
         sink._add(value);
       }
     } catch (e, s) {
       // If either _expand or iterating the generated iterator throws,
       // we abort the iteration.
-      sink._addError(e, s);
+      _addErrorWithReplacement(sink, e, s);
     }
   }
 }
 
 
 typedef bool _ErrorTest(error);
 
 /**
  * A stream pipe that converts or disposes error events
  * before passing them on.
  */
 class _HandleErrorStream<T> extends _ForwardingStream<T, T> {
   final Function _transform;
   final _ErrorTest _test;
 
   _HandleErrorStream(Stream<T> source,
                      Function onError,
                      bool test(error))
       : this._transform = onError, this._test = test, super(source);
 
   void _handleError(Object error, StackTrace stackTrace, _EventSink<T> sink) {
     bool matches = true;
     if (_test != null) {
       try {
         matches = _test(error);
       } catch (e, s) {
-        sink._addError(e, s);
+        _addErrorWithReplacement(sink, e, s);
         return;
       }
     }
     if (matches) {
       try {
         _invokeErrorHandler(_transform, error, stackTrace);
       } catch (e, s) {
         if (identical(e, error)) {
           sink._addError(error, stackTrace);
         } else {
-          sink._addError(e, s);
+          _addErrorWithReplacement(sink, e, s);
         }
         return;
       }
     } else {
       sink._addError(error, stackTrace);
     }
   }
 }
 
 
@@ skipping 25 matching lines @@
   final _Predicate<T> _test;
 
   _TakeWhileStream(Stream<T> source, bool test(T value))
       : this._test = test, super(source);
 
   void _handleData(T inputEvent, _EventSink<T> sink) {
     bool satisfies;
     try {
       satisfies = _test(inputEvent);
     } catch (e, s) {
-      sink._addError(e, s);
+      _addErrorWithReplacement(sink, e, s);
       // The test didn't say true. Didn't say false either, but we stop anyway.
       sink._close();
       return;
     }
     if (satisfies) {
       sink._add(inputEvent);
     } else {
       sink._close();
     }
   }
@@ skipping 27 matching lines @@
 
   void _handleData(T inputEvent, _EventSink<T> sink) {
     if (_hasFailed) {
       sink._add(inputEvent);
       return;
     }
     bool satisfies;
     try {
       satisfies = _test(inputEvent);
     } catch (e, s) {
-      sink._addError(e, s);
+      _addErrorWithReplacement(sink, e, s);
       // A failure to return a boolean is considered "not matching".
       _hasFailed = true;
       return;
     }
     if (!satisfies) {
       _hasFailed = true;
       sink._add(inputEvent);
     }
   }
 }
@@ skipping 15 matching lines @@
       return sink._add(inputEvent);
     } else {
       bool isEqual;
       try {
         if (_equals == null) {
           isEqual = (_previous == inputEvent);
         } else {
           isEqual = _equals(_previous, inputEvent);
         }
       } catch (e, s) {
-        sink._addError(e, s);
+        _addErrorWithReplacement(sink, e, s);
         return null;
       }
       if (!isEqual) {
         sink._add(inputEvent);
         _previous = inputEvent;
       }
     }
   }
 }