Adapt streams for additional stackTrace argument.

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 12 matching lines @@
   // TODO(lrn): Use the _state field on _ControllerSubscription to
   // also store this state. Requires that the subscription implementation
   // does not assume that it's use of the state integer is the only use.
   int _eventState;
 
   _BroadcastSubscriptionLink _next;
   _BroadcastSubscriptionLink _previous;
 
   _BroadcastSubscription(_StreamControllerLifecycle controller,
                          void onData(T data),
-                         void onError(Object error),
+                         Function onError,
                          void onDone(),
                          bool cancelOnError)
       : super(controller, onData, onError, onDone, cancelOnError) {
     _next = _previous = this;
   }
 
   _BroadcastStreamController get _controller => super._controller;
 
   bool _expectsEvent(int eventId) =>
       (_eventState & _STATE_EVENT_ID) == eventId;
@@ skipping 122 matching lines @@
     _BroadcastSubscriptionLink previous = subscription._previous;
     _BroadcastSubscriptionLink next = subscription._next;
     previous._next = next;
     next._previous = previous;
     subscription._next = subscription._previous = subscription;
   }
 
   // _StreamControllerLifecycle interface.
 
   StreamSubscription<T> _subscribe(void onData(T data),
-                                   void onError(Object error),
+                                   Function onError,
                                    void onDone(),
                                    bool cancelOnError) {
     if (isClosed) {
       throw new StateError("Subscribing to closed stream");
     }
     StreamSubscription subscription = new _BroadcastSubscription<T>(
         this, onData, onError, onDone, cancelOnError);
     _addListener(subscription);
     if (identical(_next, _previous)) {
       // Only one listener, so it must be the first listener.
@@ skipping 33 matching lines @@
   }
 
   void add(T data) {
     if (!_mayAddEvent) throw _addEventError();
     _sendData(data);
   }
 
   void addError(Object error, [Object stackTrace]) {
     if (!_mayAddEvent) throw _addEventError();
     if (stackTrace != null) _attachStackTrace(error, stackTrace);
-    _sendError(error);
+    _sendError(error, stackTrace);
   }
 
   Future close() {
     if (isClosed) {
       assert(_doneFuture != null);
       return _doneFuture;
     }
     if (!_mayAddEvent) throw _addEventError();
     _state |= _STATE_CLOSED;
     Future doneFuture = _ensureDoneFuture();
     _sendDone();
     return doneFuture;
   }
 
   Future get done => _ensureDoneFuture();
 
   Future addStream(Stream<T> stream) {
     if (!_mayAddEvent) throw _addEventError();
     _state |= _STATE_ADDSTREAM;
     _addStreamState = new _AddStreamState(this, stream);
     return _addStreamState.addStreamFuture;
   }
 
   // _EventSink interface, called from AddStreamState.
   void _add(T data) {
     _sendData(data);
   }
 
-  void _addError(Object error) {
+  void _addError(Object error, StackTrace stackTrace) {
     assert(_isAddingStream);
-    _sendError(error);
+    _sendError(error, stackTrace);
   }
 
   void _close() {
     assert(_isAddingStream);
     _AddStreamState addState = _addStreamState;
     _addStreamState = null;
     _state &= ~_STATE_ADDSTREAM;
     addState.complete();
   }
 
@@ skipping 54 matching lines @@
 
   // EventDispatch interface.
 
   void _sendData(T data) {
     if (_isEmpty) return;
     _forEachListener((_BufferingStreamSubscription<T> subscription) {
       subscription._add(data);
     });
   }
 
-  void _sendError(Object error) {
+  void _sendError(Object error, StackTrace stackTrace) {
     if (_isEmpty) return;
     _forEachListener((_BufferingStreamSubscription<T> subscription) {
-      subscription._addError(error);
+      subscription._addError(error, stackTrace);
     });
   }
 
   void _sendDone() {
     if (!_isEmpty) {
       _forEachListener((_BroadcastSubscription<T> subscription) {
         subscription._close();
       });
     } else {
       assert(_doneFuture != null);
@@ skipping 11 matching lines @@
 
   void _sendData(T data) {
     for (_BroadcastSubscriptionLink link = _next;
          !identical(link, this);
          link = link._next) {
       _BroadcastSubscription<T> subscription = link;
       subscription._addPending(new _DelayedData(data));
     }
   }
 
-  void _sendError(Object error) {
+  void _sendError(Object error, StackTrace stackTrace) {
     for (_BroadcastSubscriptionLink link = _next;
          !identical(link, this);
          link = link._next) {
       _BroadcastSubscription<T> subscription = link;
-      subscription._addPending(new _DelayedError(error));
+      subscription._addPending(new _DelayedError(error, stackTrace));
     }
   }
 
   void _sendDone() {
     if (!_isEmpty) {
       for (_BroadcastSubscriptionLink link = _next;
            !identical(link, this);
            link = link._next) {
         _BroadcastSubscription<T> subscription = link;
         subscription._addPending(const _DelayedDone());
@@ skipping 40 matching lines @@
       return;
     }
     super.add(data);
     while (_hasPending) {
       _pending.handleNext(this);
     }
   }
 
   void addError(Object error, [StackTrace stackTrace]) {
     if (!isClosed && _isFiring) {
-      _addPendingEvent(new _DelayedError(error));
+      _addPendingEvent(new _DelayedError(error, stackTrace));
       return;
     }
     super.addError(error, stackTrace);
     while (_hasPending) {
       _pending.handleNext(this);
     }
   }
 
   Future close() {
     if (!isClosed && _isFiring) {
@@ skipping 13 matching lines @@
     }
     super._callOnCancel();
   }
 }
 
 // A subscription that never receives any events.
 // It can simulate pauses, but otherwise does nothing.
 class _DoneSubscription<T> implements StreamSubscription<T> {
   int _pauseCount = 0;
   void onData(void handleData(T data)) {}
-  void onError(void handleErrr(Object error)) {}
+  void onError(Function handleError) {}
   void onDone(void handleDone()) {}
   void pause([Future resumeSignal]) {
     if (resumeSignal != null) resumeSignal.then(_resume);
     _pauseCount++;
   }
   void resume() { _resume(null); }
   void _resume(_) {
     if (_pauseCount > 0) _pauseCount--;
   }
   void cancel() {}
   bool get isPaused => _pauseCount > 0;
   Future asFuture([Object value]) => new _Future();
 }