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

sdk/lib/async/stream.dart

 // Copyright (c) 2013, 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;
 
 // -------------------------------------------------------------------
 // Core Stream types
 // -------------------------------------------------------------------
 
@@ skipping 66 matching lines @@
   /**
    * Creates a new single-subscription stream from the future.
    *
    * When the future completes, the stream will fire one event, either
    * data or error, and then close with a done-event.
    */
   factory Stream.fromFuture(Future<T> future) {
     // Use the controller's buffering to fill in the value even before
     // the stream has a listener. For a single value, it's not worth it
     // to wait for a listener before doing the `then` on the future.
-    StreamController<T> controller = new StreamController<T>(sync: true);
+    _StreamController<T> controller = new StreamController<T>(sync: true);
     future.then((value) {
-        controller.add(value);
-        controller.close();
+        controller._add(value);
+        controller._closeUnchecked();
       },
       onError: (error, stackTrace) {
-        controller.addError(error, stackTrace);
-        controller.close();
+        controller._addError(error, stackTrace);
+        controller._closeUnchecked();
       });
     return controller.stream;
   }
 
   /**
    * Creates a single-subscription stream that gets its data from [data].
    *
    * The iterable is iterated when the stream receives a listener, and stops
    * iterating if the listener cancels the subscription.
    *
@@ skipping 201 matching lines @@
    * This acts like [map], except that [convert] may return a [Future],
    * and in that case, the stream waits for that future to complete before
    * continuing with its result.
    *
    * The returned stream is a broadcast stream if this stream is.
    */
   Stream asyncMap(convert(T event)) {
     StreamController controller;
     StreamSubscription subscription;
     void onListen () {
-      var add = controller.add;
-      var addError = controller.addError;
+      final add = controller.add;
+      assert(controller is _StreamController ||
+             controller is _BroadcastStreamController);
+      final eventSink = controller;
+      final addError = eventSink._addError;
       subscription = this.listen(
           (T event) {
             var newValue;
             try {
               newValue = convert(event);
             } catch (e, s) {
               controller.addError(e, s);
               return;
             }
             if (newValue is Future) {
@@ skipping 36 matching lines @@
    *
    * If [convert] returns `null`, no value is put on the output stream,
    * just as if it returned an empty stream.
    *
    * The returned stream is a broadcast stream if this stream is.
    */
   Stream asyncExpand(Stream convert(T event)) {
     StreamController controller;
     StreamSubscription subscription;
     void onListen() {
+      assert(controller is _StreamController ||
+             controller is _BroadcastStreamController);
+      final eventSink = controller;
       subscription = this.listen(
           (T event) {
             Stream newStream;
             try {
               newStream = convert(event);
             } catch (e, s) {
               controller.addError(e, s);
               return;
             }
             if (newStream != null) {
               subscription.pause();
               controller.addStream(newStream)
                         .whenComplete(subscription.resume);
             }
           },
-          onError: controller.addError,
+          onError: eventSink._addError,  // Avoid Zone error replacement.
           onDone: controller.close
       );
     }
     if (this.isBroadcast) {
       controller = new StreamController.broadcast(
         onListen: onListen,
         onCancel: () { subscription.cancel(); },
         sync: true
       );
     } else {
@@ skipping 97 matching lines @@
           value = element;
           seenFirst = true;
         }
       },
       onError: result._completeError,
       onDone: () {
         if (!seenFirst) {
           try {
             throw IterableElementError.noElement();
           } catch (e, s) {
-            result._completeError(e, s);
+            _completeWithErrorCallback(result, e,  s);
           }
         } else {
           result._complete(value);
         }
       },
       cancelOnError: true
     );
     return result;
   }
 
@@ skipping 37 matching lines @@
     bool first = true;
     subscription = this.listen(
       (T element) {
         if (!first) {
           buffer.write(separator);
         }
         first = false;
         try {
           buffer.write(element);
         } catch (e, s) {
-          _cancelAndError(subscription, result, e, s);
+          _cancelAndErrorWithReplacement(subscription, result, e, s);
         }
       },
       onError: (e) {
         result._completeError(e);
       },
       onDone: () {
         result._complete(buffer.toString());
       },
       cancelOnError: true);
     return result;
@@ skipping 326 matching lines @@
     StreamSubscription subscription;
     subscription = this.listen(
       (T value) {
         _cancelAndValue(subscription, future, value);
       },
       onError: future._completeError,
       onDone: () {
         try {
           throw IterableElementError.noElement();
         } catch (e, s) {
-          future._completeError(e, s);
+          _completeWithErrorCallback(future, e, s);
         }
       },
       cancelOnError: true);
     return future;
   }
 
   /**
    * Returns the last element of the stream.
    *
    * If an error event occurs before the first data event, the resulting future
@@ skipping 14 matching lines @@
       },
       onError: future._completeError,
       onDone: () {
         if (foundResult) {
           future._complete(result);
           return;
         }
         try {
           throw IterableElementError.noElement();
         } catch (e, s) {
-          future._completeError(e, s);
+          _completeWithErrorCallback(future, e, s);
         }
       },
       cancelOnError: true);
     return future;
   }
 
   /**
    * Returns the single element.
    *
    * If an error event occurs before or after the first data event, the
    * resulting future is completed with that error.
    *
    * If [this] is empty or has more than one element throws a [StateError].
    */
   Future<T> get single {
     _Future<T> future = new _Future<T>();
     T result = null;
     bool foundResult = false;
     StreamSubscription subscription;
     subscription = this.listen(
       (T value) {
         if (foundResult) {
           // This is the second element we get.
           try {
             throw IterableElementError.tooMany();
           } catch (e, s) {
-            _cancelAndError(subscription, future, e, s);
+            _cancelAndErrorWithReplacement(subscription, future, e, s);
           }
           return;
         }
         foundResult = true;
         result = value;
       },
       onError: future._completeError,
       onDone: () {
         if (foundResult) {
           future._complete(result);
           return;
         }
         try {
           throw IterableElementError.noElement();
         } catch (e, s) {
-          future._completeError(e, s);
+          _completeWithErrorCallback(future, e, s);
         }
       },
       cancelOnError: true);
     return future;
   }
 
   /**
    * Finds the first element of this stream matching [test].
    *
    * Returns a future that is filled with the first element of this stream
@@ skipping 31 matching lines @@
       },
       onError: future._completeError,
       onDone: () {
         if (defaultValue != null) {
           _runUserCode(defaultValue, future._complete, future._completeError);
           return;
         }
         try {
           throw IterableElementError.noElement();
         } catch (e, s) {
-          future._completeError(e, s);
+          _completeWithErrorCallback(future, e, s);
         }
       },
       cancelOnError: true);
     return future;
   }
 
   /**
    * Finds the last element in this stream matching [test].
    *
    * As [firstWhere], except that the last matching element is found.
@@ skipping 24 matching lines @@
           future._complete(result);
           return;
         }
         if (defaultValue != null) {
           _runUserCode(defaultValue, future._complete, future._completeError);
           return;
         }
         try {
           throw IterableElementError.noElement();
         } catch (e, s) {
-          future._completeError(e, s);
+          _completeWithErrorCallback(future, e, s);
         }
       },
       cancelOnError: true);
     return future;
   }
 
   /**
    * Finds the single element in this stream matching [test].
    *
    * Like [lastMatch], except that it is an error if more than one
    * matching element occurs in the stream.
    */
   Future<T> singleWhere(bool test(T element)) {
     _Future<T> future = new _Future<T>();
     T result = null;
     bool foundResult = false;
     StreamSubscription subscription;
     subscription = this.listen(
       (T value) {
         _runUserCode(
           () => true == test(value),
           (bool isMatch) {
             if (isMatch) {
               if (foundResult) {
                 try {
                   throw IterableElementError.tooMany();
                 } catch (e, s) {
-                  _cancelAndError(subscription, future, e, s);
+                  _cancelAndErrorWithReplacement(subscription, future, e, s);
                 }
                 return;
               }
               foundResult = true;
               result = value;
             }
           },
           _cancelAndErrorClosure(subscription, future)
         );
       },
       onError: future._completeError,
       onDone: () {
         if (foundResult) {
           future._complete(result);
           return;
         }
         try {
           throw IterableElementError.noElement();
         } catch (e, s) {
-          future._completeError(e, s);
+          _completeWithErrorCallback(future, e, s);
         }
       },
       cancelOnError: true);
     return future;
   }
 
   /**
    * Returns the value of the [index]th data event of this stream.
    *
    * Stops listening to the stream after the [index]th data event has been
@@ skipping 59 matching lines @@
     Zone zone;
     Function timeout;
 
     void onData(T event) {
       timer.cancel();
       controller.add(event);
       timer = zone.createTimer(timeLimit, timeout);
     }
     void onError(error, StackTrace stackTrace) {
       timer.cancel();
-      controller.addError(error, stackTrace);
+      assert(controller is _StreamController ||
+             controller is _BroadcastStreamController);
+      var eventSink = controller;
+      eventSink._addError(error, stackTrace);  // Avoid Zone error replacement.
       timer = zone.createTimer(timeLimit, timeout);
     }
     void onDone() {
       timer.cancel();
       controller.close();
     }
     void onListen() {
       // This is the onListen callback for of controller.
       // It runs in the same zone that the subscription was created in.
       // Use that zone for creating timers and running the onTimeout
       // callback.
       zone = Zone.current;
       if (onTimeout == null) {
         timeout = () {
           controller.addError(new TimeoutException("No stream event",
-                                                   timeLimit));
+                                                   timeLimit), null);
         };
       } else {
         onTimeout = zone.registerUnaryCallback(onTimeout);
         _ControllerEventSinkWrapper wrapper =
             new _ControllerEventSinkWrapper(null);
         timeout = () {
           wrapper._sink = controller;  // Only valid during call.
           zone.runUnaryGuarded(onTimeout, wrapper);
           wrapper._sink = null;
         };
@@ skipping 385 matching lines @@
 class _ControllerEventSinkWrapper<T> implements EventSink<T> {
   EventSink _sink;
   _ControllerEventSinkWrapper(this._sink);
 
   void add(T data) { _sink.add(data); }
   void addError(error, [StackTrace stackTrace]) {
     _sink.addError(error, stackTrace);
   }
   void close() { _sink.close(); }
 }