Remove AsyncError with Expando.

sdk/lib/async/stream.dart

 // Copyright (c) 2013, 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;
 
 // -------------------------------------------------------------------
 // Core Stream types
 // -------------------------------------------------------------------
 
@@ skipping 152 matching lines @@
     return new _SingleStreamMultiplexer<T>(this);
   }
 
   /**
    * Adds a subscription to this stream.
    *
    * On each data event from this stream, the subscriber's [onData] handler
    * is called. If [onData] is null, nothing happens.
    *
    * On errors from this stream, the [onError] handler is given a
-   * [AsyncError] object describing the error.
+   * object describing the error.
    *
    * If this stream closes, the [onDone] handler is called.
    *
    * If [unsubscribeOnError] is true, the subscription is ended when
    * the first error is reported. The default is false.
    */
   StreamSubscription<T> listen(void onData(T event),
-                               { void onError(AsyncError error),
+                               { void onError(error),
                                  void onDone(),
                                  bool unsubscribeOnError});
 
   /**
    * Creates a new stream from this stream that discards some data events.
    *
    * The new stream sends the same error and done events as this stream,
    * but it only sends the data events that satisfy the [test].
    */
   Stream<T> where(bool test(T event)) {
@@ skipping 18 matching lines @@
    * true. If [test] is omitted, every error is considered matching.
    *
    * If the error is intercepted, the [handle] function can decide what to do
    * with it. It can throw if it wants to raise a new (or the same) error,
    * or simply return to make the stream forget the error.
    *
    * If you need to transform an error into a data event, use the more generic
    * [Stream.transformEvent] to handle the event by writing a data event to
    * the output sink
    */
-  Stream<T> handleError(void handle(AsyncError error), { bool test(error) }) {
+  Stream<T> handleError(void handle( error), { bool test(error) }) {
     return new _HandleErrorStream<T>(this, handle, test);
   }
 
   /**
    * Creates a new stream from this stream that converts each element
    * into zero or more events.
    *
    * Each incoming event is converted to an [Iterable] of new events,
    * and each of these new events are then sent by the returned stream
    * in order.
@@ skipping 35 matching lines @@
                        (T newValue) { value = newValue; },
                        _cancelAndError(subscription, result));
         } else {
           value = element;
           seenFirst = true;
         }
       },
       onError: result._setError,
       onDone: () {
         if (!seenFirst) {
-          result._setError(new AsyncError(new StateError("No elements")));
+          result._setError(new StateError("No elements"));
         } else {
           result._setValue(value);
         }
       },
       unsubscribeOnError: true
     );
     return result;
   }
 
   /** Reduces a sequence of values by repeatedly applying [combine]. */
   Future fold(var initialValue, combine(var previous, T element)) {
     _FutureImpl result = new _FutureImpl();
     var value = initialValue;
     StreamSubscription subscription;
     subscription = this.listen(
       // TODO(ahe): Restore type when feature is implemented in dart2js
       // checked mode. http://dartbug.com/7733
       (/*T*/ element) {
         _runUserCode(
           () => combine(value, element),
           (newValue) { value = newValue; },
           _cancelAndError(subscription, result)
         );
       },
-      onError: (AsyncError e) {
+      onError: (e) {
         result._setError(e);
       },
       onDone: () {
         result._setValue(value);
       },
       unsubscribeOnError: true);
     return result;
   }
 
   /**
@@ skipping 229 matching lines @@
     subscription = this.listen(
       // TODO(ahe): Restore type when feature is implemented in dart2js
       // checked mode. http://dartbug.com/7733
       (/*T*/ value) {
         subscription.cancel();
         future._setValue(value);
         return;
       },
       onError: future._setError,
       onDone: () {
-        future._setError(new AsyncError(new StateError("No elements")));
+        future._setError(new StateError("No elements"));
       },
       unsubscribeOnError: true);
     return future;
   }
 
   /**
    * Returns the last element.
    *
    * If [this] is empty throws a [StateError].
    */
   Future<T> get last {
     _FutureImpl<T> future = new _FutureImpl<T>();
     T result = null;
     bool foundResult = false;
     StreamSubscription subscription;
     subscription = this.listen(
       // TODO(ahe): Restore type when feature is implemented in dart2js
       // checked mode. http://dartbug.com/7733
       (/*T*/ value) {
         foundResult = true;
         result = value;
       },
       onError: future._setError,
       onDone: () {
         if (foundResult) {
           future._setValue(result);
           return;
         }
-        future._setError(new AsyncError(new StateError("No elements")));
+        future._setError(new StateError("No elements"));
       },
       unsubscribeOnError: true);
     return future;
   }
 
   /**
    * Returns the single element.
    *
    * If [this] is empty or has more than one element throws a [StateError].
    */
   Future<T> get single {
     _FutureImpl<T> future = new _FutureImpl<T>();
     T result = null;
     bool foundResult = false;
     StreamSubscription subscription;
     subscription = this.listen(
       // TODO(ahe): Restore type when feature is implemented in dart2js
       // checked mode. http://dartbug.com/7733
       (/*T*/ value) {
         if (foundResult) {
           subscription.cancel();
           // This is the second element we get.
           Error error = new StateError("More than one element");
-          future._setError(new AsyncError(error));
+          future._setError(error);
           return;
         }
         foundResult = true;
         result = value;
       },
       onError: future._setError,
       onDone: () {
         if (foundResult) {
           future._setValue(result);
           return;
         }
-        future._setError(new AsyncError(new StateError("No elements")));
+        future._setError(new StateError("No elements"));
       },
       unsubscribeOnError: true);
     return future;
   }
 
   /**
    * Finds the first element of this stream matching [test].
    *
    * Returns a future that is filled with the first element of this stream
    * that [test] returns true for.
@@ skipping 23 matching lines @@
           },
           _cancelAndError(subscription, future)
         );
       },
       onError: future._setError,
       onDone: () {
         if (defaultValue != null) {
           _runUserCode(defaultValue, future._setValue, future._setError);
           return;
         }
-        future._setError(
-            new AsyncError(new StateError("firstMatch ended without match")));
+        future._setError(new StateError("firstMatch ended without match"));
       },
       unsubscribeOnError: true);
     return future;
   }
 
   /**
    * Finds the last element in this stream matching [test].
    *
    * As [firstWhere], except that the last matching element is found.
    * That means that the result cannot be provided before this stream
@@ skipping 22 matching lines @@
       onError: future._setError,
       onDone: () {
         if (foundResult) {
           future._setValue(result);
           return;
         }
         if (defaultValue != null) {
           _runUserCode(defaultValue, future._setValue, future._setError);
           return;
         }
-        future._setError(
-            new AsyncError(new StateError("lastMatch ended without match")));
+        future._setError(new StateError("lastMatch ended without match"));
       },
       unsubscribeOnError: true);
     return future;
   }
 
   /**
    * Finds the single element in this stream matching [test].
    *
    * Like [lastMatch], except that it is an error if more than one
    * matching element occurs in the stream.
    */
   Future<T> singleWhere(bool test(T value)) {
     _FutureImpl<T> future = new _FutureImpl<T>();
     T result = null;
     bool foundResult = false;
     StreamSubscription subscription;
     subscription = this.listen(
       // TODO(ahe): Restore type when feature is implemented in dart2js
       // checked mode. http://dartbug.com/7733
       (/*T*/ value) {
         _runUserCode(
           () => true == test(value),
           (bool isMatch) {
             if (isMatch) {
               if (foundResult) {
                 subscription.cancel();
-                future._setError(new AsyncError(
-                    new StateError('Multiple matches for "single"')));
+                future._setError(
+                    new StateError('Multiple matches for "single"'));
                 return;
               }
               foundResult = true;
               result = value;
             }
           },
           _cancelAndError(subscription, future)
         );
       },
       onError: future._setError,
       onDone: () {
         if (foundResult) {
           future._setValue(result);
           return;
         }
-        future._setError(
-            new AsyncError(new StateError("single ended without match")));
+        future._setError(new StateError("single ended without match"));
       },
       unsubscribeOnError: true);
     return future;
   }
 
   /**
    * Returns the value of the [index]th data event of this stream.
    *
    * If an error event occurs, the future will end with this error.
    *
@@ skipping 10 matching lines @@
       (/*T*/ value) {
         if (index == 0) {
           subscription.cancel();
           future._setValue(value);
           return;
         }
         index -= 1;
       },
       onError: future._setError,
       onDone: () {
-        future._setError(new AsyncError(
-            new StateError("Not enough elements for elementAt")));
+        future._setError(new StateError("Not enough elements for elementAt"));
       },
       unsubscribeOnError: true);
     return future;
   }
 }
 
 /**
  * A control object for the subscription on a [Stream].
  *
  * When you subscribe on a [Stream] using [Stream.listen],
  * a [StreamSubscription] object is returned. This object
  * is used to later unsubscribe again, or to temporarily pause
  * the stream's events.
  */
 abstract class StreamSubscription<T> {
   /**
    * Cancels this subscription. It will no longer receive events.
    *
    * If an event is currently firing, this unsubscription will only
    * take effect after all subscribers have received the current event.
    */
   void cancel();
 
   /** Set or override the data event handler of this subscription. */
   void onData(void handleData(T data));
 
   /** Set or override the error event handler of this subscription. */
-  void onError(void handleError(AsyncError error));
+  void onError(void handleError(error));
 
   /** Set or override the done event handler of this subscription. */
   void onDone(void handleDone());
 
   /**
    * Request that the stream pauses events until further notice.
    *
    * If [resumeSignal] is provided, the stream will undo the pause
    * when the future completes. If the future completes with an error,
    * it will not be handled!
@@ skipping 26 matching lines @@
 }
 
 
 /**
  * An interface that abstracts creation or handling of [Stream] events.
  */
 abstract class EventSink<T> {
   /** Create a data event */
   void add(T event);
   /** Create an async error. */
-  void addError(AsyncError errorEvent);
+  void addError(errorEvent);
   /** Request a stream to close. */
   void close();
 }
 
 
 /** [Stream] wrapper that only exposes the [Stream] interface. */
 class StreamView<T> extends Stream<T> {
   Stream<T> _stream;
 
   StreamView(this._stream);
 
   bool get isBroadcast => _stream.isBroadcast;
 
   Stream<T> asBroadcastStream() => _stream.asBroadcastStream();
 
   StreamSubscription<T> listen(void onData(T value),
-                               { void onError(AsyncError error),
+                               { void onError(error),
                                  void onDone(),
                                  bool unsubscribeOnError }) {
     return _stream.listen(onData, onError: onError, onDone: onDone,
                           unsubscribeOnError: unsubscribeOnError);
   }
 }
 
 /**
  * [EventSink] wrapper that only exposes the [EventSink] interface.
  */
 class _EventSinkView<T> extends EventSink<T> {
   final EventSink<T> _sink;
 
   _EventSinkView(this._sink);
 
   void add(T value) { _sink.add(value); }
-  void addError(AsyncError error) { _sink.addError(error); }
+  void addError(error) { _sink.addError(error); }
   void close() { _sink.close(); }
 }
 
 
 /**
  * The target of a [Stream.pipe] call.
  *
  * The [Stream.pipe] call will pass itself to this object, and then return
  * the resulting [Future]. The pipe should complete the future when it's
  * done.
@@ skipping 58 matching lines @@
    *
    *     stringStream.transform(new StreamTransformer<String, String>(
    *         handleData: (Strung value, EventSink<String> sink) {
    *           sink.add(value);
    *           sink.add(value);  // Duplicate the incoming events.
    *         }));
    *
    */
   factory StreamTransformer({
       void handleData(S data, EventSink<T> sink),
-      void handleError(AsyncError error, EventSink<T> sink),
+      void handleError(error, EventSink<T> sink),
       void handleDone(EventSink<T> sink)}) {
     return new _StreamTransformerImpl<S, T>(handleData,
                                             handleError,
                                             handleDone);
   }
 
   Stream<T> bind(Stream<S> stream);
 }
 
 
@@ skipping 63 matching lines @@
     var data = event;
     sink.add(data);
   }
 
   /**
    * Act on incoming error event.
    *
    * The method may generate any number of events on the sink, but should
    * not throw.
    */
-  void handleError(AsyncError error, EventSink<T> sink) {
+  void handleError(error, EventSink<T> sink) {
     sink.addError(error);
   }
 
   /**
    * Act on incoming done event.
    *
    * The method may generate any number of events on the sink, but should
    * not throw.
    */
   void handleDone(EventSink<T> sink){
@@ skipping 11 matching lines @@
  * events on this stream.
  */
 class EventTransformStream<S, T> extends Stream<T> {
   final Stream<S> _source;
   final StreamEventTransformer _transformer;
   EventTransformStream(Stream<S> source,
                        StreamEventTransformer<S, T> transformer)
       : _source = source, _transformer = transformer;
 
   StreamSubscription<T> listen(void onData(T data),
-                               { void onError(AsyncError error),
+                               { void onError(error),
                                  void onDone(),
                                  bool unsubscribeOnError }) {
     unsubscribeOnError = identical(true, unsubscribeOnError);
     return new _EventTransformStreamSubscription(_source, _transformer,
                                                  onData, onError, onDone,
                                                  unsubscribeOnError);
   }
 }
 
 class _EventTransformStreamSubscription<S, T>
     extends _BaseStreamSubscription<T>
     implements _EventOutputSink<T> {
   /** The transformer used to transform events. */
   final StreamEventTransformer<S, T> _transformer;
   /** Whether to unsubscribe when emitting an error. */
   final bool _unsubscribeOnError;
   /** Whether this stream has sent a done event. */
   bool _isClosed = false;
   /** Source of incoming events. */
   StreamSubscription<S> _subscription;
   /** Cached EventSink wrapper for this class. */
   EventSink<T> _sink;
 
   _EventTransformStreamSubscription(Stream<S> source,
                                     this._transformer,
                                     void onData(T data),
-                                    void onError(AsyncError error),
+                                    void onError(error),
                                     void onDone(),
                                     this._unsubscribeOnError)
       : super(onData, onError, onDone) {
     _sink = new _EventOutputSinkWrapper<T>(this);
     _subscription = source.listen(_handleData,
                                   onError: _handleError,
                                   onDone: _handleDone);
   }
 
   /** Whether this subscription is still subscribed to its source. */
@@ skipping 17 matching lines @@
   }
 
   void _handleData(S data) {
     try {
       _transformer.handleData(data, _sink);
     } catch (e, s) {
       _sendError(_asyncError(e, s));
     }
   }
 
-  void _handleError(AsyncError error) {
+  void _handleError(error) {
     try {
       _transformer.handleError(error, _sink);
     } catch (e, s) {
-      _sendError(_asyncError(e, s, error));
+      _sendError(_asyncError(e, s));
     }
   }
 
   void _handleDone() {
     try {
       _subscription = null;
       _transformer.handleDone(_sink);
     } catch (e, s) {
       _sendError(_asyncError(e, s));
     }
   }
 
   // EventOutputSink interface.
   void _sendData(T data) {
     if (_isClosed) return;
     _onData(data);
   }
 
-  void _sendError(AsyncError error) {
+  void _sendError(error) {
     if (_isClosed) return;
     _onError(error);
     if (_unsubscribeOnError) {
       cancel();
     }
   }
 
   void _sendDone() {
     if (_isClosed) throw new StateError("Already closed.");
     _isClosed = true;
     if (_isSubscribed) {
       _subscription.cancel();
       _subscription = null;
     }
     _onDone();
   }
 }
 
 class _EventOutputSinkWrapper<T> extends EventSink<T> {
   _EventOutputSink _sink;
   _EventOutputSinkWrapper(this._sink);
 
   void add(T data) { _sink._sendData(data); }
-  void addError(AsyncError error) { _sink._sendError(error); }
+  void addError(error) { _sink._sendError(error); }
   void close() { _sink._sendDone(); }
 }