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

sdk/lib/async/stream_pipe.dart

Index: sdk/lib/async/stream_pipe.dart
diff --git a/sdk/lib/async/stream_pipe.dart b/sdk/lib/async/stream_pipe.dart
index 4be169c0d9fb2b1d46376fd4e020a50be24ec20d..309a771607167c4f1b4007dde538a170630a995f 100644
--- a/sdk/lib/async/stream_pipe.dart
+++ b/sdk/lib/async/stream_pipe.dart
@@ -11,7 +11,12 @@ _runUserCode(userCode(),
   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);
+    }
   }
 }
 
@@ -29,6 +34,17 @@ void _cancelAndError(StreamSubscription subscription,
   }
 }
 
+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(
@@ -169,6 +185,16 @@ class _ForwardingStreamSubscription<S, T>
 
 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;
 
@@ -180,7 +206,7 @@ class _WhereStream<T> extends _ForwardingStream<T, T> {
     try {
       satisfies = _test(inputEvent);
     } catch (e, s) {
-      sink._addError(e, s);
+      _addErrorWithReplacement(sink, e, s);
       return;
     }
     if (satisfies) {
@@ -206,7 +232,7 @@ class _MapStream<S, T> extends _ForwardingStream<S, T> {
     try {
       outputEvent = _transform(inputEvent);
     } catch (e, s) {
-      sink._addError(e, s);
+      _addErrorWithReplacement(sink, e, s);
       return;
     }
     sink._add(outputEvent);
@@ -230,7 +256,7 @@ class _ExpandStream<S, T> extends _ForwardingStream<S, T> {
     } catch (e, s) {
       // If either _expand or iterating the generated iterator throws,
       // we abort the iteration.
-      sink._addError(e, s);
+      _addErrorWithReplacement(sink, e, s);
     }
   }
 }
@@ -257,7 +283,7 @@ class _HandleErrorStream<T> extends _ForwardingStream<T, T> {
       try {
         matches = _test(error);
       } catch (e, s) {
-        sink._addError(e, s);
+        _addErrorWithReplacement(sink, e, s);
         return;
       }
     }
@@ -268,7 +294,7 @@ class _HandleErrorStream<T> extends _ForwardingStream<T, T> {
         if (identical(e, error)) {
           sink._addError(error, stackTrace);
         } else {
-          sink._addError(e, s);
+          _addErrorWithReplacement(sink, e, s);
         }
         return;
       }
@@ -314,7 +340,7 @@ class _TakeWhileStream<T> extends _ForwardingStream<T, T> {
     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;
@@ -362,7 +388,7 @@ class _SkipWhileStream<T> extends _ForwardingStream<T, T> {
     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;
@@ -398,7 +424,7 @@ class _DistinctStream<T> extends _ForwardingStream<T, T> {
           isEqual = _equals(_previous, inputEvent);
         }
       } catch (e, s) {
-        sink._addError(e, s);
+        _addErrorWithReplacement(sink, e, s);
         return null;
       }
       if (!isEqual) {