Add Stream.forEach

sdk/lib/async/stream.dart

Index: sdk/lib/async/stream.dart
diff --git a/sdk/lib/async/stream.dart b/sdk/lib/async/stream.dart
index 62174c86608657bbecec343747a67de1685c1c06..341626587b4dad9d56158de3ab6b19fe2b16509e 100644
--- a/sdk/lib/async/stream.dart
+++ b/sdk/lib/async/stream.dart
@@ -339,6 +339,34 @@ abstract class Stream<T> {
   }
 
   /**
+   * Executes [action] on each data event of the stream.
+   *
+   * Completes the returned [Future] when all events of the stream
+   * have been processed. Completes the future with an error if the
+   * stream has an error event, or if [action] throws.
+   */
+  Future forEach(void action(T element)) {
+    _FutureImpl future = new _FutureImpl();
+    StreamSubscription subscription;
+    subscription = this.listen(
+        // TODO(ahe): Restore type when feature is implemented in dart2js
+        // checked mode. http://dartbug.com/7733
+        (/*T*/ element) {
+          _runUserCode(
+            () => action(element),
+            (_) {},
+            _cancelAndError(subscription, future)
+          );
+        },
+        onError: future._setError,
+        onDone: () {
+          future._setValue(null);
+        },
+        cancelOnError: true);
+    return future;
+  }
+
+  /**
    * Checks whether [test] accepts all elements provided by this stream.
    *
    * Completes the [Future] when the answer is known.