Add Stream.forEach

sdk/lib/async/stream.dart

 // Copyright (c) 2013, 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 321 matching lines @@
         },
         onError: future._setError,
         onDone: () {
           future._setValue(false);
         },
         cancelOnError: true);
     return future;
   }
 
   /**
+   * 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.
    * 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(
         // TODO(ahe): Restore type when feature is implemented in dart2js
@@ skipping 825 matching lines @@
 }
 
 class _EventOutputSinkWrapper<T> extends EventSink<T> {
   _EventOutputSink _sink;
   _EventOutputSinkWrapper(this._sink);
 
   void add(T data) { _sink._sendData(data); }
   void addError(error) { _sink._sendError(error); }
   void close() { _sink._sendDone(); }
 }