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

sdk/lib/async/broadcast_stream_controller.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;
 
 class _BroadcastStream<T> extends _ControllerStream<T> {
   _BroadcastStream(_StreamControllerLifecycle controller) : super(controller);
 
   bool get isBroadcast => true;
@@ skipping 221 matching lines @@
     return new StateError("Cannot add new events while doing an addStream");
   }
 
   void add(T data) {
     if (!_mayAddEvent) throw _addEventError();
     _sendData(data);
   }
 
   void addError(Object error, [StackTrace stackTrace]) {
     if (!_mayAddEvent) throw _addEventError();
+    AsyncError replacement = Zone.current.errorCallback(error, stackTrace);
+    if (replacement != null) {
+      error = replacement.error;
+      stackTrace = replacement.stackTrace;
+    }
     _sendError(error, stackTrace);
   }
 
   Future close() {
     if (isClosed) {
       assert(_doneFuture != null);
       return _doneFuture;
     }
     if (!_mayAddEvent) throw _addEventError();
     _state |= _STATE_CLOSED;
@@ skipping 10 matching lines @@
     _addStreamState = new _AddStreamState(this, stream, cancelOnError);
     return _addStreamState.addStreamFuture;
   }
 
   // _EventSink interface, called from AddStreamState.
   void _add(T data) {
     _sendData(data);
   }
 
   void _addError(Object error, StackTrace stackTrace) {
-    assert(_isAddingStream);
     _sendError(error, stackTrace);
   }
 
   void _close() {
     assert(_isAddingStream);
     _AddStreamState addState = _addStreamState;
     _addStreamState = null;
     _state &= ~_STATE_ADDSTREAM;
     addState.complete();
   }
@@ skipping 169 matching lines @@
     while (_hasPending) {
       _pending.handleNext(this);
     }
   }
 
   void addError(Object error, [StackTrace stackTrace]) {
     if (!isClosed && _isFiring) {
       _addPendingEvent(new _DelayedError(error, stackTrace));
       return;
     }
-    super.addError(error, stackTrace);
+    if (!_mayAddEvent) throw _addEventError();
+    _sendError(error, stackTrace);
     while (_hasPending) {
       _pending.handleNext(this);
     }
   }
 
   Future close() {
     if (!isClosed && _isFiring) {
       _addPendingEvent(const _DelayedDone());
       _state |= _BroadcastStreamController._STATE_CLOSED;
       return super.done;
@@ skipping 24 matching lines @@
     _pauseCount++;
   }
   void resume() { _resume(null); }
   void _resume(_) {
     if (_pauseCount > 0) _pauseCount--;
   }
   Future cancel() { return new _Future.immediate(null); }
   bool get isPaused => _pauseCount > 0;
   Future asFuture([Object value]) => new _Future();
 }