Adapt streams for additional stackTrace argument.

sdk/lib/async/stream_impl.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;
 
 /** Abstract and private interface for a place to put events. */
 abstract class _EventSink<T> {
   void _add(T data);
-  void _addError(Object error);
+  void _addError(Object error, StackTrace stackTrace);
   void _close();
 }
 
 /**
  * Abstract and private interface for a place to send events.
  *
  * Used by event buffering to finally dispatch the pending event, where
  * [_EventSink] is where the event first enters the stream subscription,
  * and may yet be buffered.
  */
 abstract class _EventDispatch<T> {
   void _sendData(T data);
-  void _sendError(Object error);
+  void _sendError(Object error, StackTrace stackTrace);
   void _sendDone();
 }
 
 /**
  * Default implementation of stream subscription of buffering events.
  *
  * The only public methods are those of [StreamSubscription], so instances of
  * [_BufferingStreamSubscription] can be returned directly as a
  * [StreamSubscription] without exposing internal functionality.
  *
@@ skipping 37 matching lines @@
    * when `cancelOnError` is true.
    */
   static const int _STATE_CANCELED = 8;
   static const int _STATE_IN_CALLBACK = 16;
   static const int _STATE_HAS_PENDING = 32;
   static const int _STATE_PAUSE_COUNT = 64;
   static const int _STATE_PAUSE_COUNT_SHIFT = 6;
 
   /* Event handlers provided in constructor. */
   _DataHandler<T> _onData;
-  _ErrorHandler _onError;
+  Function _onError;
   _DoneHandler _onDone;
   final Zone _zone = Zone.current;
 
   /** Bit vector based on state-constants above. */
   int _state;
 
   /**
    * Queue of pending events.
    *
    * Is created when necessary, or set in constructor for preconfigured events.
    */
   _PendingEvents _pending;
 
   _BufferingStreamSubscription(void onData(T data),
-                               void onError(error),
+                               Function onError,
                                void onDone(),
                                bool cancelOnError)
       : _onData = Zone.current.registerUnaryCallback(onData),
-        _onError = Zone.current.registerUnaryCallback(onError),
+        _onError = _registerErrorCallback(onError),
         _onDone = Zone.current.registerCallback(onDone),
         _state = (cancelOnError ? _STATE_CANCEL_ON_ERROR : 0) {
     assert(_onData != null);
     assert(_onError != null);
     assert(_onDone != null);
   }
 
+  static _registerErrorCallback(Function errorCallback) {
+    if (errorCallback is ZoneBinaryCallback) {
+      return Zone.current.registerBinaryCallback(errorCallback);
+    } else {
+      return Zone.current.registerUnaryCallback(errorCallback);
+    }
+  }
+
   /**
    * Sets the subscription's pending events object.
    *
    * This can only be done once. The pending events object is used for the
    * rest of the subscription's life cycle.
    */
   void _setPendingEvents(_PendingEvents pendingEvents) {
     assert(_pending == null);
     if (pendingEvents == null) return;
     _pending = pendingEvents;
@@ skipping 13 matching lines @@
     assert(_isCanceled);
     _PendingEvents events = _pending;
     _pending = null;
     return events;
   }
 
   // StreamSubscription interface.
 
   void onData(void handleData(T event)) {
     if (handleData == null) handleData = _nullDataHandler;
-    _onData = handleData;
+    _onData = Zone.current.registerUnaryCallback(handleData);
   }
 
-  void onError(void handleError(error)) {
+  void onError(Function handleError) {
     if (handleError == null) handleError = _nullErrorHandler;
-    _onError = handleError;
+    _onError = _registerErrorCallback(handleError);
   }
 
   void onDone(void handleDone()) {
     if (handleDone == null) handleDone = _nullDoneHandler;
-    _onDone = handleDone;
+    _onDone = Zone.current.registerCallback(handleDone);
   }
 
   void pause([Future resumeSignal]) {
     if (_isCanceled) return;
     bool wasPaused = _isPaused;
     bool wasInputPaused = _isInputPaused;
     // Increment pause count and mark input paused (if it isn't already).
     _state = (_state + _STATE_PAUSE_COUNT) | _STATE_INPUT_PAUSED;
     if (resumeSignal != null) resumeSignal.whenComplete(resume);
     if (!wasPaused && _pending != null) _pending.cancelSchedule();
@@ skipping 27 matching lines @@
       _pending = null;
       _state &= ~_STATE_IN_CALLBACK;
     }
   }
 
   Future asFuture([var futureValue]) {
     _Future<T> result = new _Future<T>();
 
     // Overwrite the onDone and onError handlers.
     _onDone = () { result._complete(futureValue); };
-    _onError = (error) {
+    _onError = (error, stackTrace) {
       cancel();
-      result._completeError(error);
+      result._completeError(error, stackTrace);
     };
 
     return result;
   }
 
   // State management.
 
   bool get _isInputPaused => (_state & _STATE_INPUT_PAUSED) != 0;
   bool get _isClosed => (_state & _STATE_CLOSED) != 0;
   bool get _isCanceled => (_state & _STATE_CANCELED) != 0;
@@ skipping 40 matching lines @@
   void _add(T data) {
     assert(!_isClosed);
     if (_isCanceled) return;
     if (_canFire) {
       _sendData(data);
     } else {
       _addPending(new _DelayedData(data));
     }
   }
 
-  void _addError(Object error) {
+  void _addError(Object error, StackTrace stackTrace) {
     if (_isCanceled) return;
     if (_canFire) {
-      _sendError(error);  // Reports cancel after sending.
+      _sendError(error, stackTrace);  // Reports cancel after sending.
     } else {
-      _addPending(new _DelayedError(error));
+      _addPending(new _DelayedError(error, stackTrace));
     }
   }
 
   void _close() {
     assert(!_isClosed);
     if (_isCanceled) return;
     _state |= _STATE_CLOSED;
     if (_canFire) {
       _sendDone();
     } else {
@@ skipping 43 matching lines @@
     assert(!_isCanceled);
     assert(!_isPaused);
     assert(!_inCallback);
     bool wasInputPaused = _isInputPaused;
     _state |= _STATE_IN_CALLBACK;
     _zone.runUnaryGuarded(_onData, data);
     _state &= ~_STATE_IN_CALLBACK;
     _checkState(wasInputPaused);
   }
 
-  void _sendError(var error) {
+  void _sendError(var error, StackTrace stackTrace) {
     assert(!_isCanceled);
     assert(!_isPaused);
     assert(!_inCallback);
     bool wasInputPaused = _isInputPaused;
     _state |= _STATE_IN_CALLBACK;
     if (!_zone.inSameErrorZone(Zone.current)) {
       // Errors are not allowed to traverse zone boundaries.
-      Zone.current.handleUncaughtError(error, getAttachedStackTrace(error));
+      Zone.current.handleUncaughtError(error, stackTrace);
+    } else if (_onError is ZoneBinaryCallback) {
+      _zone.runBinaryGuarded(_onError, error, stackTrace);
     } else {
       _zone.runUnaryGuarded(_onError, error);
     }
     _state &= ~_STATE_IN_CALLBACK;
     if (_cancelOnError) {
       _cancel();
     }
     _checkState(wasInputPaused);
   }
 
@@ skipping 67 matching lines @@
 }
 
 // -------------------------------------------------------------------
 // Common base class for single and multi-subscription streams.
 // -------------------------------------------------------------------
 abstract class _StreamImpl<T> extends Stream<T> {
   // ------------------------------------------------------------------
   // Stream interface.
 
   StreamSubscription<T> listen(void onData(T data),
-                               { void onError(error),
+                               { Function onError,
                                  void onDone(),
                                  bool cancelOnError }) {
     if (onData == null) onData = _nullDataHandler;
     if (onError == null) onError = _nullErrorHandler;
     if (onDone == null) onDone = _nullDoneHandler;
     cancelOnError = identical(true, cancelOnError);
     StreamSubscription subscription =
         _createSubscription(onData, onError, onDone, cancelOnError);
     _onListen(subscription);
     return subscription;
@@ skipping 21 matching lines @@
   final _EventGenerator _pending;
   /**
    * Initializes the stream to have only the events provided by a
    * [_PendingEvents].
    *
    * A new [_PendingEvents] must be generated for each listen.
    */
   _GeneratedStreamImpl(this._pending);
 
   StreamSubscription _createSubscription(void onData(T data),
-                                         void onError(Object error),
+                                         Function onError,
                                          void onDone(),
                                          bool cancelOnError) {
     _BufferingStreamSubscription<T> subscription =
          new _BufferingStreamSubscription(
              onData, onError, onDone, cancelOnError);
     subscription._setPendingEvents(_pending());
     return subscription;
   }
 }
 
@@ skipping 15 matching lines @@
     // Send one event per call to moveNext.
     // If moveNext returns true, send the current element as data.
     // If moveNext returns false, send a done event and clear the _iterator.
     // If moveNext throws an error, send an error and clear the _iterator.
     // After an error, no further events will be sent.
     bool isDone;
     try {
       isDone = !_iterator.moveNext();
     } catch (e, s) {
       _iterator = null;
-      dispatch._sendError(_asyncError(e, s));
+      dispatch._sendError(_asyncError(e, s), s);
       return;
     }
     if (!isDone) {
       dispatch._sendData(_iterator.current);
     } else {
       _iterator = null;
       dispatch._sendDone();
     }
   }
 
   void clear() {
     if (isScheduled) cancelSchedule();
     _iterator = null;
   }
 }
 
 
 // Internal helpers.
 
 // Types of the different handlers on a stream. Types used to type fields.
 typedef void _DataHandler<T>(T value);
-typedef void _ErrorHandler(error);
 typedef void _DoneHandler();
 
 
 /** Default data handler, does nothing. */
 void _nullDataHandler(var value) {}
 
 /** Default error handler, reports the error to the current zone's handler. */
-void _nullErrorHandler(error) {
-  Zone.current.handleUncaughtError(error, getAttachedStackTrace(error));
+void _nullErrorHandler(error, [StackTrace stackTrace]) {
+  Zone.current.handleUncaughtError(error, stackTrace);
 }
 
 /** Default done handler, does nothing. */
 void _nullDoneHandler() {}
 
 
 /** A delayed event on a buffering stream subscription. */
 abstract class _DelayedEvent {
   /** Added as a linked list on the [StreamController]. */
   _DelayedEvent next;
   /** Execute the delayed event on the [StreamController]. */
   void perform(_EventDispatch dispatch);
 }
 
 /** A delayed data event. */
 class _DelayedData<T> extends _DelayedEvent {
   final T value;
   _DelayedData(this.value);
   void perform(_EventDispatch<T> dispatch) {
     dispatch._sendData(value);
   }
 }
 
 /** A delayed error event. */
 class _DelayedError extends _DelayedEvent {
   final error;
-  _DelayedError(this.error);
+  final StackTrace stackTrace;
+
+  _DelayedError(this.error, this.stackTrace);
   void perform(_EventDispatch dispatch) {
-    dispatch._sendError(error);
+    dispatch._sendError(error, stackTrace);
   }
 }
 
 /** A delayed done event. */
 class _DelayedDone implements _DelayedEvent {
   const _DelayedDone();
   void perform(_EventDispatch dispatch) {
     dispatch._sendDone();
   }
 
@@ skipping 121 matching lines @@
 
 typedef void _broadcastCallback(StreamSubscription subscription);
 
 /**
  * Dummy subscription that will never receive any events.
  */
 class _DummyStreamSubscription<T> implements StreamSubscription<T> {
   int _pauseCounter = 0;
 
   void onData(void handleData(T data)) {}
-  void onError(void handleError(Object data)) {}
+  void onError(Function handleError) {}
   void onDone(void handleDone()) {}
 
   void pause([Future resumeSignal]) {
     _pauseCounter++;
     if (resumeSignal != null) resumeSignal.then((_) { resume(); });
   }
   void resume() {
     if (_pauseCounter > 0) _pauseCounter--;
   }
   void cancel() {}
@@ skipping 16 matching lines @@
                      void onCancelHandler(StreamSubscription subscription))
       : _onListenHandler = Zone.current.registerUnaryCallback(onListenHandler),
         _onCancelHandler = Zone.current.registerUnaryCallback(onCancelHandler),
         _zone = Zone.current {
     _controller = new _AsBroadcastStreamController<T>(_onListen, _onCancel);
   }
 
   bool get isBroadcast => true;
 
   StreamSubscription<T> listen(void onData(T data),
-                               { void onError(Object error),
+                               { Function onError,
                                  void onDone(),
                                  bool cancelOnError}) {
     if (_controller == null) {
       // Return a dummy subscription backed by nothing, since
       // it won't ever receive any events.
       return new _DummyStreamSubscription<T>();
     }
     if (_subscription == null) {
       _subscription = _source.listen(_controller.add,
                                      onError: _controller.addError,
@@ skipping 214 matching lines @@
       _state = _STATE_FOUND;
       hasNext._complete(true);
       return;
     }
     _subscription.pause();
     assert(_futureOrPrefetch == null);
     _futureOrPrefetch = data;
     _state = _STATE_EXTRA_DATA;
   }
 
-  void _onError(Object error) {
+  void _onError(Object error, [StackTrace stackTrace]) {
     if (_state == _STATE_MOVING) {
       _Future<bool> hasNext = _futureOrPrefetch;
       // We have cancelOnError: true, so the subscription is canceled.
       _clear();
-      hasNext._completeError(error);
+      hasNext._completeError(error, stackTrace);
       return;
     }
     _subscription.pause();
     assert(_futureOrPrefetch == null);
     _futureOrPrefetch = error;
     _state = _STATE_EXTRA_ERROR;
   }
 
   void _onDone() {
      if (_state == _STATE_MOVING) {
       _Future<bool> hasNext = _futureOrPrefetch;
       _clear();
       hasNext._complete(false);
       return;
     }
     _subscription.pause();
     _futureOrPrefetch = null;
     _state = _STATE_EXTRA_DONE;
   }
 }