Make Streams also cosider a thrown AsyncError a rethrow.

sdk/lib/async/stream_pipe.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;
 
+/** Utility function to create an [AsyncError] if [error] isn't one already. */
+AsyncError _asyncError(Object error, Object stackTrace, [AsyncError cause]) {
+  if (error is AsyncError) return error;
+  if (cause == null) return new AsyncError(error, stackTrace);
+  return new AsyncError.withCause(error, stackTrace, cause);
+}
+
+/** Runs user code and takes actions depending on success or failure. */
+_runUserCode(userCode(), onSuccess(value), onError(AsyncError error),
+             { AsyncError cause }) {
+  var result;
+  try {
+    result = userCode();
+  } on AsyncError catch (e) {
+    return onError(e);
+  } catch (e, s) {
+    if (cause == null) {
+      onError(new AsyncError(e, s));
+    } else {
+      onError(new AsyncError.withCause(e, s, cause));
+    }
+  }
+  onSuccess(result);
+}
+
+/** Helper function to make an onError argument to [_runUserCode]. */
+_cancelAndError(StreamSubscription subscription, _FutureImpl future) =>
+  (AsyncError error) {
+    subscription.cancel();
+    future._setError(error);
+  };
+
+
 /**
  * A wrapper around a stream that allows independent subscribers.
  *
  * By default [this] subscribes to [_source] and forwards all events to its own
  * subscribers. It does not subscribe until there is a subscriber, and
  * unsubscribes again when there are no subscribers left.
  *
  * The events are passed through the [_handleData], [_handleError] and
  * [_handleDone] methods. Subclasses are supposed to add handling of some of
  * the events by overriding these methods.
@@ skipping 51 matching lines @@
 
   void _handleDone() {
     _close();
   }
 }
 
 
 abstract class _ForwardingTransformer<S, T> extends _ForwardingMultiStream<S, T>
                                             implements StreamTransformer<S, T> {
   Stream<T> bind(Stream<S> source) {
-    assert(_source == null);
+    if (_source != null) throw new StateError("Already bound to source.");
     _source = source;
     if (_hasSubscribers) {
       _subscribeToSource();
     }
     return this;
   }
 }
 
 // -------------------------------------------------------------------
 // Stream transformers used by the default Stream implementation.
 // -------------------------------------------------------------------
 
 typedef bool _Predicate<T>(T value);
 
 class WhereStream<T> extends _ForwardingTransformer<T, T> {
   final _Predicate<T> _test;
 
   WhereStream(bool test(T value))
       : this._test = test;
 
   void _handleData(T inputEvent) {
     bool satisfies;
     try {
       satisfies = _test(inputEvent);
     } catch (e, s) {
-      _signalError(new AsyncError(e, s));
+      _signalError(_asyncError(e, s));
       return;
     }
     if (satisfies) {
       _add(inputEvent);
     }
   }
 }
 
 
 typedef T _Transformation<S, T>(S value);
 
 /**
  * A stream pipe that converts data events before passing them on.
  */
 class MapStream<S, T> extends _ForwardingTransformer<S, T> {
   final _Transformation _transform;
 
   MapStream(T transform(S event))
       : this._transform = transform;
 
   void _handleData(S inputEvent) {
     T outputEvent;
     try {
       outputEvent = _transform(inputEvent);
     } catch (e, s) {
-      _signalError(new AsyncError(e, s));
+      _signalError(_asyncError(e, s));
       return;
     }
     _add(outputEvent);
   }
 }
 
 /**
  * A stream pipe that converts data events before passing them on.
  */
 class ExpandStream<S, T> extends _ForwardingTransformer<S, T> {
   final _Transformation<S, Iterable<T>> _expand;
 
   ExpandStream(Iterable<T> expand(S event))
       : this._expand = expand;
 
   void _handleData(S inputEvent) {
     try {
       for (T value in _expand(inputEvent)) {
         _add(value);
       }
     } catch (e, s) {
       // If either _expand or iterating the generated iterator throws,
       // we abort the iteration.
-      _signalError(new AsyncError(e, s));
+      _signalError(_asyncError(e, s));
     }
   }
 }
 
 
-typedef AsyncError _ErrorTransformation(AsyncError error);
+typedef void _ErrorTransformation(AsyncError error);
+typedef bool _ErrorTest(error);
 
 /**
  * A stream pipe that converts or disposes error events
  * before passing them on.
  */
 class HandleErrorStream<T> extends _ForwardingTransformer<T, T> {
   final _ErrorTransformation _transform;
+  final _ErrorTest _test;
 
-  HandleErrorStream(AsyncError transform(AsyncError event))
-      : this._transform = transform;
+  HandleErrorStream(void transform(AsyncError event), bool test(error))
+      : this._transform = transform, this._test = test;
 
   void _handleError(AsyncError error) {
-    try {
-      error = _transform(error);
-      if (error == null) return;
-    } catch (e, s) {
-      error = new AsyncError.withCause(e, s, error);
+    bool matches = true;
+    if (_test != null) {
+      try {
+        matches = _test(error.error);
+      } catch (e, s) {
+        _signalError(_asyncError(e, s, error));
+        return;
+      }
     }
-    _signalError(error);
+    if (matches) {
+      try {
+        _transform(error);
+      } catch (e, s) {
+        _signalError(_asyncError(e, s, error));
+        return;
+      }
+    } else {
+      _signalError(error);
+    }
   }
 }
 
 
 typedef void _TransformDataHandler<S, T>(S data, StreamSink<T> sink);
 typedef void _TransformErrorHandler<T>(AsyncError data, StreamSink<T> sink);
 typedef void _TransformDoneHandler<T>(StreamSink<T> sink);
 
 /**
  * A stream pipe that intercepts all events and can generate any event as
@@ skipping 17 matching lines @@
         this._onError = (onError == null ? _defaultHandleError : onError),
         this._onDone = (onDone == null ? _defaultHandleDone : onDone) {
     // Cache the sink wrapper to avoid creating a new one for each event.
     this._sink = new _StreamImplSink(this);
   }
 
   void _handleData(S data) {
     try {
       return _onData(data, _sink);
     } catch (e, s) {
-      _signalError(new AsyncError(e, s));
+      _signalError(_asyncError(e, s));
     }
   }
 
   void _handleError(AsyncError error) {
     try {
       _onError(error, _sink);
     } catch (e, s) {
-      _signalError(new AsyncError.withCause(e, s, error));
+      _signalError(_asyncError(e, s, error));
     }
   }
 
   void _handleDone() {
     try {
       _onDone(_sink);
     } catch (e, s) {
-      _signalError(new AsyncError(e, s));
+      _signalError(_asyncError(e, s));
     }
   }
 
   /** Default data handler forwards all data. */
   static void _defaultHandleData(dynamic data, StreamSink sink) {
     sink.add(data);
   }
   /** Default error handler forwards all errors. */
   static void _defaultHandleError(AsyncError error, StreamSink sink) {
     sink.signalError(error);
@@ skipping 29 matching lines @@
   TransformStream(StreamTransformer<S, T> transform)
       : this._transform = transform {
     // Cache the sink wrapper to avoid creating a new one for each event.
     this._sink = new _StreamImplSink(this);
   }
 
   void _handleData(S data) {
     try {
       return _transform.handleData(data, _sink);
     } catch (e, s) {
-      _controller.signalError(new AsyncError(e, s));
+      _controller.signalError(_asyncError(e, s));
     }
   }
 
   void _handleError(AsyncError error) {
     try {
       _transform.handleError(error, _sink);
     } catch (e, s) {
-      _controller.signalError(new AsyncError.withCause(e, s, error));
+      _controller.signalError(_asyncError(e, s, error));
     }
   }
 
   void _handleDone() {
     try {
       _transform.handleDone(_sink);
     } catch (e, s) {
-      _controller.signalError(new AsyncError(e, s));
+      _controller.signalError(_asyncError(e, s));
     }
   }
 }
 
 
 /** Helper class for transforming three functions into a StreamTransformer. */
 class _StreamTransformerFunctionWrapper<S, T>
     extends _StreamTransformer<S, T> {
   final _TransformDataHandler<S, T> _handleData;
   final _TransformErrorHandler<T> _handleError;
@@ skipping 50 matching lines @@
   final _Predicate<T> _test;
 
   TakeWhileStream(bool test(T value))
       : this._test = test;
 
   void _handleData(T inputEvent) {
     bool satisfies;
     try {
       satisfies = _test(inputEvent);
     } catch (e, s) {
-      _signalError(new AsyncError(e, s));
+      _signalError(_asyncError(e, s));
       // The test didn't say true. Didn't say false either, but we stop anyway.
       _close();
       return;
     }
     if (satisfies) {
       _add(inputEvent);
     } else {
       _close();
     }
   }
@@ skipping 26 matching lines @@
       : this._test = test;
 
   void _handleData(T inputEvent) {
     if (_hasFailed) {
       _add(inputEvent);
     }
     bool satisfies;
     try {
       satisfies = _test(inputEvent);
     } catch (e, s) {
-      _signalError(new AsyncError(e, s));
+      _signalError(_asyncError(e, s));
       // A failure to return a boolean is considered "not matching".
       _hasFailed = true;
       return;
     }
     if (!satisfies) {
       _hasFailed = true;
       _add(inputEvent);
     }
   }
 }
@@ skipping 15 matching lines @@
       return _add(inputEvent);
     } else {
       bool isEqual;
       try {
         if (_equals == null) {
           isEqual = (_previous == inputEvent);
         } else {
           isEqual = _equals(_previous, inputEvent);
         }
       } catch (e, s) {
-        _signalError(new AsyncError(e, s));
+        _signalError(_asyncError(e, s));
         return null;
       }
       if (!isEqual) {
         _add(inputEvent);
         _previous = inputEvent;
       }
     }
   }
 }