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, evert error is considered mathcing.

[floitsch, 2013/01/11 13:23:33]

every ... matching

[Lasse Reichstein Nielsen, 2013/01/14 08:33:30]

Done.

+   *
+   * 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 20 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);
+        } on AsyncError catch (e) {
+          subscription.cancel();
+          result._setError(e);
         } catch (e, s) {
           subscription.cancel();
           result._setError(new AsyncError(e, s));
         }
       },
       onError: (AsyncError e) {
         result._setError(e);
       },
       onDone: () {
         result._setValue(value);
@@ skipping 23 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) {
+          bool matches;
+          try {
+            matches = (element == match);
+          } on AsyncError catch (e) {

[floitsch, 2013/01/11 13:23:33]

This is repeated too often...
runUserCode(userCode, then, future);

runUserCode(() => element == match,
            (bool matches) { if (matches) ... },
            future);

[Lasse Reichstein Nielsen, 2013/01/14 08:33:30]

Done.

It's not pretty, but it's a little smaller and less boilerplate.

+            subscription.cancel();
+            future._setError(e);
+            return;
+          } catch (e, s) {
+            subscription.cancel();
+            future._setError(new AsyncError(e, s));
+            return;
+          }
+          if (matches) {
             subscription.cancel();
             future._setValue(true);
           }
         },
         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)) {
+          bool matches;
+          try {
+            matches = test(element);
+          } on AsyncError catch (e) {
+            subscription.cancel();
+            future._setError(e);
+            return;
+          } catch (e, s) {
+            subscription.cancel();
+            future._setError(new AsyncError(e, s));
+            return;
+          }
+          if (!matches) {
             subscription.cancel();
             future._setValue(false);
           }
         },
         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)) {
+          bool matches;
+          try {
+            matches = test(element);
+          } on AsyncError catch (e) {
+            subscription.cancel();
+            future._setError(e);
+            return;
+          } catch (e, s) {
+            subscription.cancel();
+            future._setError(new AsyncError(e, s));
+            return;
+          }
+          if (matches) {
             subscription.cancel();
             future._setValue(true);
           }
         },
         onError: future._setError,
         onDone: () {
           future._setValue(false);
         },
         unsubscribeOnError: true);
     return future;
@@ skipping 26 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;
+          bool foundSmaller;
+          try {
+            foundSmaller = (compare(min, value) > 0);
+          } on AsyncError catch (e) {
+            subscription.cancel();
+            future._setError(e);
+            return;
+          } catch (e, s) {
+            subscription.cancel();
+            future._setError(new AsyncError(e, s));
+            return;
+          }
+          if (foundSmaller) min = value;
         });
       },
       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;
+          bool foundGreater;
+          try {
+            foundGreater = (compare(min, value) < 0);
+          } on AsyncError catch (e) {
+            subscription.cancel();
+            future._setError(e);
+            return;
+          } catch (e, s) {
+            subscription.cancel();
+            future._setError(new AsyncError(e, s));
+            return;
+          }
+          if (foundGreater) min = value;
         });
       },
       onError: future._setError,
       onDone: () {
         future._setValue(max);
       },
       unsubscribeOnError: true
     );
     return future;
   }
@@ skipping 203 matching lines @@
    * 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));
+        } on AsyncError catch (e) {
+          subscription.cancel();
+          future._setError(e);
+          return;
         } catch (e, s) {
+          subscription.cancel();
           future._setError(new AsyncError(e, s));
-          subscription.cancel();
           return;
         }
         if (matches) {
+          subscription.cancel();
           future._setValue(value);
-          subscription.cancel();
         }
       },
       onError: future._setError,
       onDone: () {
         if (defaultValue != null) {
           T value;
           try {
             value = defaultValue();
+          } on AsyncError catch (e) {
+            future._setError(e);
+            return;
           } catch (e, s) {
             future._setError(new AsyncError(e, s));
             return;
           }
           future._setValue(value);
           return;
         }
         future._setError(
             new AsyncError(new StateError("firstMatch ended without match")));
       },
@@ skipping 11 matching lines @@
   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));
+        } on AsyncError catch (e) {
+          subscription.cancel();
+          future._setError(e);
+          return;
         } catch (e, s) {
+          subscription.cancel();
           future._setError(new AsyncError(e, s));
-          subscription.cancel();
           return;
         }
         if (matches) {
           foundResult = true;
           result = value;
         }
       },
       onError: future._setError,
       onDone: () {
         if (foundResult) {
           future._setValue(result);
           return;
         }
         if (defaultValue != null) {
           T value;
           try {
             value = defaultValue();
+          } on AsyncError catch (e) {
+            future._setError(e);
+            return;
           } catch (e, s) {
             future._setError(new AsyncError(e, s));
             return;
           }
           future._setValue(value);
           return;
         }
         future._setError(
             new AsyncError(new StateError("lastMatch ended without match")));
       },
@@ skipping 10 matching lines @@
   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));
+        } on AsyncError catch (e) {
+          subscription.cancel();
+          future._setError(e);
+          return;
         } catch (e, s) {
+          subscription.cancel();
           future._setError(new AsyncError(e, s));
-          subscription.cancel();
           return;
         }
         if (matches) {
           if (foundResult) {
+            subscription.cancel();
             future._setError(new AsyncError(
                 new StateError('Multiple matches for "single"')));
-            subscription.cancel();
             return;
           }
           foundResult = true;
           result = value;
         }
       },
       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 (0 > index) throw new ArgumentError(index);

[floitsch, 2013/01/11 13:23:33]

No. If you don't want to check for int, don't check, but don't invert the order.

[Lasse Reichstein Nielsen, 2013/01/14 08:33:30]

Done.

     _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();
   }
 }