Make Streams also cosider a thrown AsyncError a rethrow.

sdk/lib/async/stream.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;
 
 // -------------------------------------------------------------------
 // Core Stream types
 // -------------------------------------------------------------------
 
@@ skipping 92 matching lines @@
    * Create a new stream that converts each element of this stream
    * to a new value using the [convert] function.
    */
   Stream mappedBy(convert(T event)) {
     return this.transform(new MapStream<T, dynamic>(convert));
   }
 
   /**
    * Create a wrapper Stream that intercepts some errors from this stream.
    *
-   * If the handler returns null, the error is considered handled.
-   * Otherwise the returned [AsyncError] is passed to the subscribers
-   * of the stream.
+   * If this stream sends an error that matches [test], then it is intercepted
+   * by the [handle] function.
+   *
+   * An [AsyncError] [:e:] is matched by a test function if [:test(e):] returns
+   * 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.
    */
-  Stream handleError(AsyncError handle(AsyncError error)) {
-    return this.transform(new HandleErrorStream<T>(handle));
+  Stream<T> handleError(void handle(AsyncError error), { bool test(error) }) {
+    return this.transform(new HandleErrorStream<T>(handle, test));
   }
 
   /**
    * Create 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 18 matching lines @@
   }
 
 
   /** Reduces a sequence of values by repeatedly applying [combine]. */
   Future reduce(var initialValue, combine(var previous, T element)) {
     _FutureImpl result = new _FutureImpl();
     var value = initialValue;
     StreamSubscription subscription;
     subscription = this.listen(
       (T element) {
-        try {
-          value = combine(value, element);
-        } catch (e, s) {
-          subscription.cancel();
-          result._setError(new AsyncError(e, s));
-        }
+        _runUserCode(
+          () => combine(value, element),
+          (result) { value = result; },
+          _cancelAndError(subscription, result)
+        );
       },
       onError: (AsyncError e) {
         result._setError(e);
       },
       onDone: () {
         result._setValue(value);
       },
       unsubscribeOnError: true);
     return result;
   }
@@ skipping 19 matching lines @@
    * Check whether [match] occurs in the elements provided by this stream.
    *
    * Completes the [Future] when the answer is known.
    * If this stream reports an error, the [Future] will report that error.
    */
   Future<bool> contains(T match) {
     _FutureImpl<bool> future = new _FutureImpl<bool>();
     StreamSubscription subscription;
     subscription = this.listen(
         (T element) {
-          if (element == match) {
-            subscription.cancel();
-            future._setValue(true);
-          }
+          _runUserCode(
+            () => match(element),
+            (bool isMatch) {
+              if (isMatch) {
+                subscription.cancel();
+                future._setValue(element);
+              }
+            },
+            _cancelAndError(subscription, future)
+          );
         },
         onError: future._setError,
         onDone: () {
           future._setValue(false);
         },
         unsubscribeOnError: true);
     return future;
   }
 
   /**
    * Check whether [test] accepts all elements provided by this stream.
    *
    * Completes the [Future] when the answer is known.
    * If this stream reports an error, the [Future] will report that error.
    */
   Future<bool> every(bool test(T element)) {
     _FutureImpl<bool> future = new _FutureImpl<bool>();
     StreamSubscription subscription;
     subscription = this.listen(
         (T element) {
-          if (!test(element)) {
-            subscription.cancel();
-            future._setValue(false);
-          }
+          _runUserCode(
+            () => test(element),
+            (bool isMatch) {
+              if (!isMatch) {
+                subscription.cancel();
+                future._setValue(false);
+              }
+            },
+            _cancelAndError(subscription, future)
+          );
         },
         onError: future._setError,
         onDone: () {
           future._setValue(true);
         },
         unsubscribeOnError: true);
     return future;
   }
 
   /**
    * Check whether [test] accepts any element provided by this stream.
    *
    * Completes the [Future] when the answer is known.
    * If this stream reports an error, the [Future] will report that error.
    */
   Future<bool> any(bool test(T element)) {
     _FutureImpl<bool> future = new _FutureImpl<bool>();
     StreamSubscription subscription;
     subscription = this.listen(
         (T element) {
-          if (test(element)) {
-            subscription.cancel();
-            future._setValue(true);
-          }
+          _runUserCode(
+            () => test(element),
+            (bool isMatch) {
+              if (isMatch) {
+                subscription.cancel();
+                future._setValue(true);
+              }
+            },
+            _cancelAndError(subscription, future)
+          );
         },
         onError: future._setError,
         onDone: () {
           future._setValue(false);
         },
         unsubscribeOnError: true);
     return future;
   }
 
 
@@ skipping 23 matching lines @@
    */
   Future<T> min([int compare(T a, T b)]) {
     if (compare == null) compare = Comparable.compare;
     _FutureImpl<T> future = new _FutureImpl<T>();
     StreamSubscription subscription;
     T min = null;
     subscription = this.listen(
       (T value) {
         min = value;
         subscription.onData((T value) {
-          if (compare(min, value) > 0) min = value;
+          _runUserCode(
+            () => compare(min, value) > 0,
+            (bool foundSmaller) {
+              if (foundSmaller) {
+                min = value;
+              }
+            },
+            _cancelAndError(subscription, future)
+          );
         });
       },
       onError: future._setError,
       onDone: () {
         future._setValue(min);
       },
       unsubscribeOnError: true
     );
     return future;
   }
@@ skipping 10 matching lines @@
    */
   Future<T> max([int compare(T a, T b)]) {
     if (compare == null) compare = Comparable.compare;
     _FutureImpl<T> future = new _FutureImpl<T>();
     StreamSubscription subscription;
     T max = null;
     subscription = this.listen(
       (T value) {
         max = value;
         subscription.onData((T value) {
-          if (compare(max, value) < 0) max = value;
+          _runUserCode(
+            () => compare(max, value) < 0,
+            (bool foundGreater) {
+              if (foundGreater) {
+                max = value;
+              }
+            },
+            _cancelAndError(subscription, future)
+          );
         });
       },
       onError: future._setError,
       onDone: () {
         future._setValue(max);
       },
       unsubscribeOnError: true
     );
     return future;
   }
@@ skipping 200 matching lines @@
    *
    * If an error occurs, or if this stream ends without finding a match and
    * with no [defaultValue] function provided, the future will receive an
    * error.
    */
   Future<T> firstMatching(bool test(T value), {T defaultValue()}) {
     _FutureImpl<T> future = new _FutureImpl<T>();
     StreamSubscription subscription;
     subscription = this.listen(
       (T value) {
-        bool matches;
-        try {
-          matches = (true == test(value));
-        } catch (e, s) {
-          future._setError(new AsyncError(e, s));
-          subscription.cancel();
-          return;
-        }
-        if (matches) {
-          future._setValue(value);
-          subscription.cancel();
-        }
+        _runUserCode(
+          () => test(value),
+          (bool isMatch) {
+            if (isMatch) {
+              subscription.cancel();
+              future._setValue(value);
+            }
+          },
+          _cancelAndError(subscription, future)
+        );
       },
       onError: future._setError,
       onDone: () {
         if (defaultValue != null) {
-          T value;
-          try {
-            value = defaultValue();
-          } catch (e, s) {
-            future._setError(new AsyncError(e, s));
-            return;
-          }
-          future._setValue(value);
+          _runUserCode(defaultValue, future._setValue, future._setError);
           return;
         }
         future._setError(
             new AsyncError(new StateError("firstMatch ended without match")));
       },
       unsubscribeOnError: true);
     return future;
   }
 
   /**
    * Finds the last element in this stream matching [test].
    *
    * As [firstMatching], except that the last matching element is found.
    * That means that the result cannot be provided before this stream
    * is done.
    */
   Future<T> lastMatching(bool test(T value), {T defaultValue()}) {
     _FutureImpl<T> future = new _FutureImpl<T>();
     T result = null;
     bool foundResult = false;
     StreamSubscription subscription;
     subscription = this.listen(
       (T value) {
-        bool matches;
-        try {
-          matches = (true == test(value));
-        } catch (e, s) {
-          future._setError(new AsyncError(e, s));
-          subscription.cancel();
-          return;
-        }
-        if (matches) {
-          foundResult = true;
-          result = value;
-        }
+        _runUserCode(
+          () => true == test(value),
+          (bool isMatch) {
+            if (isMatch) {
+              foundResult = true;
+              result = value;
+            }
+          },
+          _cancelAndError(subscription, future)
+        );
       },
       onError: future._setError,
       onDone: () {
         if (foundResult) {
           future._setValue(result);
           return;
         }
         if (defaultValue != null) {
-          T value;
-          try {
-            value = defaultValue();
-          } catch (e, s) {
-            future._setError(new AsyncError(e, s));
-            return;
-          }
-          future._setValue(value);
+          _runUserCode(defaultValue, future._setValue, future._setError);
           return;
         }
         future._setError(
             new AsyncError(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> singleMatching(bool test(T value)) {
     _FutureImpl<T> future = new _FutureImpl<T>();
     T result = null;
     bool foundResult = false;
     StreamSubscription subscription;
     subscription = this.listen(
       (T value) {
-        bool matches;
-        try {
-          matches = (true == test(value));
-        } catch (e, s) {
-          future._setError(new AsyncError(e, s));
-          subscription.cancel();
-          return;
-        }
-        if (matches) {
-          if (foundResult) {
-            future._setError(new AsyncError(
-                new StateError('Multiple matches for "single"')));
-            subscription.cancel();
-            return;
-          }
-          foundResult = true;
-          result = value;
-        }
+        _runUserCode(
+          () => true == test(value),
+          (bool isMatch) {
+            if (isMatch) {
+              if (foundResult) {
+                subscription.cancel();
+                future._setError(new AsyncError(
+                    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")));
       },
       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.
    *
    * If this stream provides fewer than [index] elements before closing,
    * an error is reported.
    */
   Future<T> elementAt(int index) {
+    if (index is! int || index < 0) throw new ArgumentError(index);
     _FutureImpl<T> future = new _FutureImpl();
     StreamSubscription subscription;
     subscription = this.listen(
       (T value) {
         if (index == 0) {
           future._setValue(value);
           subscription.cancel();
           return;
         }
         index -= 1;
@@ skipping 155 matching lines @@
     sink.signalError(error);
   }
 
   /**
    * Handle an incoming done event.
    */
   void handleDone(StreamSink<T> sink) {
     sink.close();
   }
 }