Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(3)

Issues Search
    My Issues | Starred     Open | Closed | All

Unified Diff: sdk/lib/async/stream.dart

Issue 11862008: Make Streams also cosider a thrown AsyncError a rethrow. (Closed) Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart
Patch Set: Fix indentation. Created 7 years, 11 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View side-by-side diff with in-line comments
Download patch
« no previous file with comments | « no previous file | sdk/lib/async/stream_controller.dart » ('j') | sdk/lib/async/stream_pipe.dart » ('J')
Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
Index: sdk/lib/async/stream.dart
diff --git a/sdk/lib/async/stream.dart b/sdk/lib/async/stream.dart
index c9377d5da2abd4d6aca92a760132e9f41de6a7df..27d06c2645810c2dd1e1e385f447aaec6e205a63 100644
--- a/sdk/lib/async/stream.dart
+++ b/sdk/lib/async/stream.dart
@@ -110,12 +110,18 @@ abstract class Stream<T> {
/**
* 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));
}
/**
@@ -156,6 +162,9 @@ abstract class Stream<T> {
(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));
@@ -199,7 +208,19 @@ abstract class Stream<T> {
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
Lasse Reichstein Nielsen 2013/01/14 08:33:30 Done. It's not pretty, but it's a little smaller
+ 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);
}
@@ -223,7 +244,19 @@ abstract class Stream<T> {
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);
}
@@ -247,7 +280,19 @@ abstract class Stream<T> {
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);
}
@@ -294,7 +339,19 @@ abstract class Stream<T> {
(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,
@@ -325,7 +382,19 @@ abstract class Stream<T> {
(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,
@@ -549,14 +618,18 @@ abstract class Stream<T> {
bool matches;
try {
matches = (true == test(value));
+ } on AsyncError catch (e) {
+ subscription.cancel();
+ future._setError(e);
+ return;
} catch (e, s) {
- future._setError(new AsyncError(e, s));
subscription.cancel();
+ future._setError(new AsyncError(e, s));
return;
}
if (matches) {
- future._setValue(value);
subscription.cancel();
+ future._setValue(value);
}
},
onError: future._setError,
@@ -565,6 +638,9 @@ abstract class Stream<T> {
T value;
try {
value = defaultValue();
+ } on AsyncError catch (e) {
+ future._setError(e);
+ return;
} catch (e, s) {
future._setError(new AsyncError(e, s));
return;
@@ -596,9 +672,13 @@ abstract class Stream<T> {
bool matches;
try {
matches = (true == test(value));
+ } on AsyncError catch (e) {
+ subscription.cancel();
+ future._setError(e);
+ return;
} catch (e, s) {
- future._setError(new AsyncError(e, s));
subscription.cancel();
+ future._setError(new AsyncError(e, s));
return;
}
if (matches) {
@@ -616,6 +696,9 @@ abstract class Stream<T> {
T value;
try {
value = defaultValue();
+ } on AsyncError catch (e) {
+ future._setError(e);
+ return;
} catch (e, s) {
future._setError(new AsyncError(e, s));
return;
@@ -646,16 +729,20 @@ abstract class Stream<T> {
bool matches;
try {
matches = (true == test(value));
+ } on AsyncError catch (e) {
+ subscription.cancel();
+ future._setError(e);
+ return;
} catch (e, s) {
- future._setError(new AsyncError(e, s));
subscription.cancel();
+ future._setError(new AsyncError(e, s));
return;
}
if (matches) {
if (foundResult) {
+ subscription.cancel();
future._setError(new AsyncError(
new StateError('Multiple matches for "single"')));
- subscription.cancel();
return;
}
foundResult = true;
@@ -684,6 +771,7 @@ abstract class Stream<T> {
* 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 chec
Lasse Reichstein Nielsen 2013/01/14 08:33:30 Done.
_FutureImpl<T> future = new _FutureImpl();
StreamSubscription subscription;
subscription = this.listen(
« no previous file with comments | « no previous file | sdk/lib/async/stream_controller.dart » ('j') | sdk/lib/async/stream_pipe.dart » ('J')

Powered by Google App Engine
This is Rietveld 408576698