Adapt streams for additional stackTrace argument.

sdk/lib/async/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;
 
 // -------------------------------------------------------------------
 // Controller for creating and adding events to a stream.
 // -------------------------------------------------------------------
 
@@ skipping 152 matching lines @@
    *
    * Also allows an objection stack trace object, on top of what [EventSink]
    * allows.
    */
   void addError(Object error, [Object stackTrace]);
 }
 
 
 abstract class _StreamControllerLifecycle<T> {
   StreamSubscription<T> _subscribe(void onData(T data),
-                                   void onError(Object error),
+                                   Function onError,
                                    void onDone(),
                                    bool cancelOnError);
   void _recordPause(StreamSubscription<T> subscription) {}
   void _recordResume(StreamSubscription<T> subscription) {}
   void _recordCancel(StreamSubscription<T> subscription) {}
 }
 
 /**
  * Default implementation of [StreamController].
  *
@@ skipping 192 matching lines @@
   /**
    * Send or enqueue an error event.
    */
   void addError(Object error, [Object stackTrace]) {
     if (!_mayAddEvent) throw _badEventState();
     if (stackTrace != null) {
       // Force stack trace overwrite. Even if the error already contained
       // a stack trace.
       _attachStackTrace(error, stackTrace);
     }
-    _addError(error);
+    _addError(error, stackTrace);
   }
 
   /**
    * Closes this controller.
    *
    * After closing, no further events may be added using [add] or [addError].
    *
    * You are allowed to close the controller more than once, but only the first
    * call has any effect.
    *
@@ skipping 20 matching lines @@
 
   // Add data event, used both by the [addStream] events and by [add].
   void _add(T value) {
     if (hasListener) {
       _sendData(value);
     } else if (_isInitialState) {
       _ensurePendingEvents().add(new _DelayedData<T>(value));
     }
   }
 
-  void _addError(Object error) {
+  void _addError(Object error, StackTrace stackTrace) {
     if (hasListener) {
-      _sendError(error);
+      _sendError(error, stackTrace);
     } else if (_isInitialState) {
-      _ensurePendingEvents().add(new _DelayedError(error));
+      _ensurePendingEvents().add(new _DelayedError(error, stackTrace));
     }
   }
 
   void _close() {
     // End of addStream stream.
     assert(_isAddingStream);
     _StreamControllerAddStreamState addState = _varData;
     _varData = addState.varData;
     _state &= ~_STATE_ADDSTREAM;
     addState.complete();
   }
 
   // _StreamControllerLifeCycle interface
 
   StreamSubscription<T> _subscribe(void onData(T data),
-                                   void onError(Object error),
+                                   Function onError,
                                    void onDone(),
                                    bool cancelOnError) {
     if (!_isInitialState) {
       throw new StateError("Stream has already been listened to.");
     }
     _ControllerSubscription subscription = new _ControllerSubscription(
         this, onData, onError, onDone, cancelOnError);
 
     _PendingEvents pendingEvents = _pendingEvents;
     _state |= _STATE_SUBSCRIBED;
@@ skipping 41 matching lines @@
     _runGuarded(_onResume);
   }
 }
 
 abstract class _SyncStreamControllerDispatch<T>
     implements _StreamController<T> {
   void _sendData(T data) {
     _subscription._add(data);
   }
 
-  void _sendError(Object error) {
-    _subscription._addError(error);
+  void _sendError(Object error, StackTrace stackTrace) {
+    _subscription._addError(error, stackTrace);
   }
 
   void _sendDone() {
     _subscription._close();
   }
 }
 
 abstract class _AsyncStreamControllerDispatch<T>
     implements _StreamController<T> {
   void _sendData(T data) {
     _subscription._addPending(new _DelayedData(data));
   }
 
-  void _sendError(Object error) {
-    _subscription._addPending(new _DelayedError(error));
+  void _sendError(Object error, StackTrace stackTrace) {
+    _subscription._addPending(new _DelayedError(error, stackTrace));
   }
 
   void _sendDone() {
     _subscription._addPending(const _DelayedDone());
   }
 }
 
 // TODO(lrn): Use common superclass for callback-controllers when VM supports
 // constructors in mixin superclasses.
 
@@ skipping 47 matching lines @@
   }
 }
 
 class _ControllerStream<T> extends _StreamImpl<T> {
   _StreamControllerLifecycle<T> _controller;
 
   _ControllerStream(this._controller);
 
   StreamSubscription<T> _createSubscription(
       void onData(T data),
-      void onError(Object error),
+      Function onError,
       void onDone(),
       bool cancelOnError) =>
     _controller._subscribe(onData, onError, onDone, cancelOnError);
 
   // Override == and hashCode so that new streams returned by the same
   // controller are considered equal. The controller returns a new stream
   // each time it's queried, but doesn't have to cache the result.
 
   int get hashCode => _controller.hashCode ^ 0x35323532;
 
   bool operator==(Object other) {
     if (identical(this, other)) return true;
     if (other is! _ControllerStream) return false;
     _ControllerStream otherStream = other;
     return identical(otherStream._controller, this._controller);
   }
 }
 
 class _ControllerSubscription<T> extends _BufferingStreamSubscription<T> {
   final _StreamControllerLifecycle<T> _controller;
 
   _ControllerSubscription(this._controller,
                           void onData(T data),
-                          void onError(Object error),
+                          Function onError,
                           void onDone(),
                           bool cancelOnError)
       : super(onData, onError, onDone, cancelOnError);
 
   void _onCancel() {
     _controller._recordCancel(this);
   }
 
   void _onPause() {
     _controller._recordPause(this);
   }
 
   void _onResume() {
     _controller._recordResume(this);
   }
 }
 
 
 /** A class that exposes only the [StreamSink] interface of an object. */
 class _StreamSinkWrapper<T> implements StreamSink<T> {
   final StreamSink _target;
   _StreamSinkWrapper(this._target);
   void add(T data) { _target.add(data); }
-  void addError(Object error) { _target.addError(error); }
+  void addError(Object error, [StackTrace stackTrace]) {
+    _target.addError(error);
+  }
   Future close() => _target.close();
   Future addStream(Stream<T> source) => _target.addStream(source);
   Future get done => _target.done;
 }
 
 /**
  * Object containing the state used to handle [StreamController.addStream].
  */
 class _AddStreamState<T> {
   // [_FutureImpl] returned by call to addStream.
@@ skipping 34 matching lines @@
   var varData;
 
   _StreamControllerAddStreamState(_StreamController controller,
                                   this.varData,
                                   Stream source) : super(controller, source) {
     if (controller.isPaused) {
       addSubscription.pause();
     }
   }
 }