Add missing null-tests to async error functions.

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) {
     AsyncError replacement = Zone.current.errorCallback(e, s);
     if (replacement == null) {
       onError(e, s);
     } else {
-      onError(replacement.error, replacement.stackTrace);
+      var error = _nonNullError(replacement.error);
+      var stackTrace = replacement.stackTrace;
+      onError(error, 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;
+    error = _nonNullError(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));
 
@@ skipping 127 matching lines @@
 }
 
 // -------------------------------------------------------------------
 // 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);
+  if (replacement != null) {
+    error = _nonNullError(replacement.error);
+    stackTrace = replacement.stackTrace;
   }
+  sink._addError(error, 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) {
@@ skipping 222 matching lines @@
         _addErrorWithReplacement(sink, e, s);
         return null;
       }
       if (!isEqual) {
         sink._add(inputEvent);
         _previous = inputEvent;
       }
     }
   }
 }