Add missing null-tests to async error functions.

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 309a771607167c4f1b4007dde538a170630a995f..f35c658d03cabc1899f0ab3e5b952f1581e37652 100644
--- a/sdk/lib/async/stream_pipe.dart
+++ b/sdk/lib/async/stream_pipe.dart
@@ -15,7 +15,10 @@ _runUserCode(userCode(),
     if (replacement == null) {
       onError(e, s);
     } else {
-      onError(replacement.error, replacement.stackTrace);
+      var error = replacement.error;
+      if (error == null) error = new NullThrownError();
+      var stackTrace = replacement.stackTrace;
+      onError(error, stackTrace);
     }
   }
 }
@@ -40,6 +43,7 @@ void _cancelAndErrorWithReplacement(StreamSubscription subscription,
   AsyncError replacement = Zone.current.errorCallback(error, stackTrace);
   if (replacement != null) {
     error = replacement.error;
+    if (error == null) error = new NullThrownError();
     stackTrace = replacement.stackTrace;
   }
   _cancelAndError(subscription, future, error, stackTrace);
@@ -187,11 +191,12 @@ 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);
+  if (replacement != null) {
+    error = replacement.error;
+    if (error == null) error = new NullThrownError();
+    stackTrace = replacement.stackTrace;
   }
+  sink._addError(error, stackTrace);
 }