Remove AsyncError with Expando.

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;
 
 // States shared by single/multi stream implementations.
 
 // Completion state of the stream.
 /// Initial and default state where the stream can receive and send events.
@@ skipping 37 matching lines @@
 const int _LISTENER_PENDING_UNSUBSCRIBE = 2;
 
 /// Bit that contains the last sent event's "id bit".
 const int _LISTENER_EVENT_ID = 4;
 const int _LISTENER_EVENT_ID_SHIFT = 2;
 
 /// The count of times a listener has paused is stored in the
 /// state, shifted by this amount.
 const int _LISTENER_PAUSE_COUNT_SHIFT = 3;
 
+/** Throws the given error in the next cycle. */
+_throwDelayed(var error, [Object stackTrace]) {
+  // We are going to reach the top-level here, but there might be a global
+  // exception handler. This means that we shouldn't print the stack trace.
+  // TODO(floitsch): Find better solution that doesn't print the stack trace
+  // if there is a global exception handler.
+  runAsync(() {
+    if (stackTrace != null) print(stackTrace);
+    var trace = getAttachedStackTrace(error);
+    if (trace != null && trace != stackTrace) print(trace);
+    throw error;
+  });
+}
+
 
 // -------------------------------------------------------------------
 // Common base class for single and multi-subscription streams.
 // -------------------------------------------------------------------
 abstract class _StreamImpl<T> extends Stream<T> {
   /** Current state of the stream. */
   int _state = _STREAM_OPEN;
 
   /**
    * List of pending events.
    *
    * If events are added to the stream (using [_add], [_addError] or [_done])
    * while the stream is paused, or while another event is firing, events will
    * stored here.
    * Also supports scheduling the events for later execution.
    */
   _PendingEvents _pendingEvents;
 
   // ------------------------------------------------------------------
   // Stream interface.
 
   StreamSubscription<T> listen(void onData(T data),
-                               { void onError(AsyncError error),
+                               { void onError(error),
                                  void onDone(),
                                  bool cancelOnError }) {
     if (_isComplete) {
       return new _DoneSubscription(onDone);
     }
     if (onData == null) onData = _nullDataHandler;
     if (onError == null) onError = _nullErrorHandler;
     if (onDone == null) onDone = _nullDoneHandler;
     cancelOnError = identical(true, cancelOnError);
     _StreamSubscriptionImpl subscription =
@@ skipping 25 matching lines @@
     }
     _handlePendingEvents();
   }
 
   /**
    * Send or enqueue an error event.
    *
    * If a subscription has requested to be unsubscribed on errors,
    * it will be unsubscribed after receiving this event.
    */
-  void _addError(AsyncError error) {
+  void _addError(error) {
     if (_isClosed) throw new StateError("Sending on closed stream");
     if (!_mayFireState) {
       // Not the time to send events.
       _addPendingEvent(new _DelayedError(error));
       return;
     }
     if (_hasPendingEvent) {
       _addPendingEvent(new _DelayedError(error));
     } else {
       _sendError(error);
@@ skipping 230 matching lines @@
 
   /** Schedule pending events to be executed. */
   void _schedulePendingEvents() {
     assert(_hasPendingEvent);
     _pendingEvents.schedule(this);
   }
 
   /** Create a subscription object. Called by [subcribe]. */
   _StreamSubscriptionImpl<T> _createSubscription(
       void onData(T data),
-      void onError(AsyncError error),
+      void onError(error),
       void onDone(),
       bool cancelOnError);
 
   /**
    * Adds a listener to this stream.
    */
   void _addListener(_StreamSubscriptionImpl subscription);
 
   /**
    * Handle a cancel requested from a [_StreamSubscriptionImpl].
@@ skipping 98 matching lines @@
   /**
    * Send a data event directly to each subscriber.
    */
   _sendData(T value) {
     assert(!_isPaused);
     assert(!_isComplete);
     if (!_hasListener) return;
     _forEachSubscriber((subscriber) {
       try {
         subscriber._sendData(value);
-      } on AsyncError catch (e) {
-        e.throwDelayed();
       } catch (e, s) {
-        new AsyncError(e, s).throwDelayed();
+        _throwDelayed(e, s);
       }
     });
   }
 
   /**
    * Sends an error event directly to each subscriber.
    */
-  void _sendError(AsyncError error) {
+  void _sendError(error) {
     assert(!_isPaused);
     assert(!_isComplete);
     if (!_hasListener) return;
     _forEachSubscriber((subscriber) {
       try {
         subscriber._sendError(error);
-      } on AsyncError catch (e) {
-        e.throwDelayed();
       } catch (e, s) {
-        new AsyncError.withCause(e, s, error).throwDelayed();
+        _throwDelayed(e, s);
       }
     });
   }
 
   /**
    * Sends the "done" message directly to each subscriber.
    * This automatically stops further subscription and
    * unsubscribes all subscribers.
    */
   void _sendDone() {
     assert(!_isPaused);
     assert(_isClosed);
     _setComplete();
     if (!_hasListener) return;
     _forEachSubscriber((subscriber) {
       _cancel(subscriber);
       try {
         subscriber._sendDone();
-      } on AsyncError catch (e) {
-        e.throwDelayed();
       } catch (e, s) {
-        new AsyncError(e, s).throwDelayed();
+        _throwDelayed(e, s);
       }
     });
     assert(!_hasListener);
   }
 }
 
 // -------------------------------------------------------------------
 // Default implementation of a stream with a single subscriber.
 // -------------------------------------------------------------------
 /**
@@ skipping 35 matching lines @@
   _SingleStreamImpl() {
     // Start out paused.
     _updatePauseCount(1);
   }
 
   /**
    * Create the new subscription object.
    */
   _StreamSubscriptionImpl<T> _createSubscription(
       void onData(T data),
-      void onError(AsyncError error),
+      void onError(error),
       void onDone(),
       bool cancelOnError) {
     return new _StreamSubscriptionImpl<T>(
         this, onData, onError, onDone, cancelOnError);
   }
 
   void _addListener(_StreamListener subscription) {
     assert(!_isComplete);
     if (_hasListener) {
       throw new StateError("Stream already has subscriber.");
@@ skipping 99 matching lines @@
   // Helper functions that can be overridden in subclasses.
 
   /** Whether there are currently any subscribers on this [Stream]. */
   bool get _hasListener => !_InternalLinkList.isEmpty(this);
 
   /**
    * Create the new subscription object.
    */
   _StreamListener<T> _createSubscription(
       void onData(T data),
-      void onError(AsyncError error),
+      void onError(error),
       void onDone(),
       bool cancelOnError) {
     return new _StreamSubscriptionImpl<T>(
         this, onData, onError, onDone, cancelOnError);
   }
 
   // -------------------------------------------------------------------
   // Internal implementation.
 
   /**
@@ skipping 108 matching lines @@
    */
   _GeneratedSingleStreamImpl(_PendingEvents events) {
     _pendingEvents = events;
     _setClosed();  // Closed for input since all events are already pending.
   }
 
   void _add(T value) {
     throw new UnsupportedError("Cannot inject events into generated stream");
   }
 
-  void _addError(AsyncError value) {
+  void _addError(value) {
     throw new UnsupportedError("Cannot inject events into generated stream");
   }
 
   void _close() {
     throw new UnsupportedError("Cannot inject events into generated stream");
   }
 }
 
 
 /** Pending events object that gets its events from an [Iterable]. */
@@ skipping 14 matching lines @@
   void handleNext(_StreamImpl<T> stream) {
     if (_isDone) throw new StateError("No events pending.");
     try {
       _isDone = !_iterator.moveNext();
       if (!_isDone) {
         stream._sendData(_iterator.current);
       } else {
         stream._sendDone();
       }
     } catch (e, s) {
-      stream._sendError(new AsyncError(e, s));
+      stream._sendError(_asyncError(e, s));
       stream._sendDone();
       _isDone = true;
     }
   }
 }
 
 
 /**
  * The subscription class that the [StreamController] uses.
  *
@@ skipping 21 matching lines @@
                           this._onData,
                           this._onError,
                           this._onDone,
                           this._cancelOnError) : super(source);
 
   void onData(void handleData(T event)) {
     if (handleData == null) handleData = _nullDataHandler;
     _onData = handleData;
   }
 
-  void onError(void handleError(AsyncError error)) {
+  void onError(void handleError(error)) {
     if (handleError == null) handleError = _nullErrorHandler;
     _onError = handleError;
   }
 
   void onDone(void handleDone()) {
     if (handleDone == null) handleDone = _nullDoneHandler;
     _onDone = handleDone;
   }
 
   void _sendData(T data) {
     _onData(data);
   }
 
-  void _sendError(AsyncError error) {
+  void _sendError(error) {
     _onError(error);
     if (_cancelOnError) _source._cancel(this);
   }
 
   void _sendDone() {
     _onDone();
   }
 
   void cancel() {
     if (!_isSubscribed) return;
     _source._cancel(this);
   }
 
   void pause([Future resumeSignal]) {
     if (!_isSubscribed) return;
     _source._pause(this, resumeSignal);
   }
 
   void resume() {
     if (!_isSubscribed || !isPaused) return;
     _source._resume(this, false);
   }
 
   Future asFuture([var futureValue]) {
     _FutureImpl<T> result = new _FutureImpl<T>();
 
     // Overwrite the onDone and onError handlers.
     onDone(() { result._setValue(futureValue); });
-    onError((AsyncError error) {
+    onError((error) {
       cancel();
       result._setError(error);
     });
 
     return result;
   }
 }
 
 // Internal helpers.
 
 // Types of the different handlers on a stream. Types used to type fields.
 typedef void _DataHandler<T>(T value);
-typedef void _ErrorHandler(AsyncError error);
+typedef void _ErrorHandler(error);
 typedef void _DoneHandler();
 
 
 /** Default data handler, does nothing. */
 void _nullDataHandler(var value) {}
 
 /** Default error handler, reports the error to the global handler. */
-void _nullErrorHandler(AsyncError error) {
-  error.throwDelayed();
+void _nullErrorHandler(error) {
+  _throwDelayed(error);
 }
 
 /** Default done handler, does nothing. */
 void _nullDoneHandler() {}
 
 
 /** A delayed event on a stream implementation. */
 abstract class _DelayedEvent {
   /** Added as a linked list on the [StreamController]. */
   _DelayedEvent next;
   /** Execute the delayed event on the [StreamController]. */
   void perform(_StreamImpl stream);
 }
 
 /** A delayed data event. */
 class _DelayedData<T> extends _DelayedEvent{
-  T value;
+  final T value;
   _DelayedData(this.value);
   void perform(_StreamImpl<T> stream) {
     stream._sendData(value);
   }
 }
 
 /** A delayed error event. */
 class _DelayedError extends _DelayedEvent {
-  AsyncError error;
+  final error;
   _DelayedError(this.error);
   void perform(_StreamImpl stream) {
     stream._sendError(error);
   }
 }
 
 /** A delayed done event. */
 class _DelayedDone implements _DelayedEvent {
   const _DelayedDone();
   void perform(_StreamImpl stream) {
@@ skipping 83 matching lines @@
     list._previousLink = otherLast;
     otherLast._nextLink = list;
     // Clean up [other].
     other._nextLink = other._previousLink = other;
   }
 }
 
 /** Abstract type for an internal interface for sending events. */
 abstract class _EventOutputSink<T> {
   _sendData(T data);
-  _sendError(AsyncError error);
+  _sendError(error);
   _sendDone();
 }
 
 abstract class _StreamListener<T> extends _InternalLink
                                   implements _EventOutputSink<T> {
   final _StreamImpl _source;
   int _state = _LISTENER_UNSUBSCRIBED;
 
   _StreamListener(this._source);
 
@@ skipping 54 matching lines @@
   void _incrementPauseCount() {
     _state += 1 << _LISTENER_PAUSE_COUNT_SHIFT;
   }
 
   void _decrementPauseCount() {
     assert(isPaused);
     _state -= 1 << _LISTENER_PAUSE_COUNT_SHIFT;
   }
 
   _sendData(T data);
-  _sendError(AsyncError error);
+  _sendError(error);
   _sendDone();
 }
 
 /** Superclass for provider of pending events. */
 abstract class _PendingEvents {
   /**
    * Timer set when pending events are scheduled for execution.
    *
    * When scheduling pending events for execution in a later cycle, the timer
    * is stored here. If pending events are executed earlier than that, e.g.,
@@ skipping 67 matching lines @@
     assert(_timer == null && _pauseCount == 0);
     _timer = new Timer(Duration.ZERO, () {
       if (_handler != null) _handler();
     });
   }
 
   bool get _isComplete => _timer == null && _pauseCount == 0;
 
   void onData(void handleAction(T value)) {}
 
-  void onError(void handleError(AsyncError error)) {}
+  void onError(void handleError(error)) {}
 
   void onDone(void handleDone()) {
     _handler = handleDone;
   }
 
   void pause([Future signal]) {
     if (_isComplete) return;
     if (_timer != null) {
       _timer.cancel();
       _timer = null;
@@ skipping 21 matching lines @@
     }
     _pauseCount = 0;
   }
 
   Future asFuture([var futureValue]) {
     // TODO(floitsch): share more code.
     _FutureImpl<T> result = new _FutureImpl<T>();
 
     // Overwrite the onDone and onError handlers.
     onDone(() { result._setValue(futureValue); });
-    onError((AsyncError error) {
+    onError((error) {
       cancel();
       result._setError(error);
     });
 
     return result;
   }
 }
 
 class _SingleStreamMultiplexer<T> extends _MultiStreamImpl<T> {
   final Stream<T> _source;
@@ skipping 24 matching lines @@
                                      onError: this._addError,
                                      onDone: this._close);
     } else {
       // TODO(lrn): Check why this can happen.
       if (_subscription == null) return;
       _subscription.cancel();
       _subscription = null;
     }
   }
 }