Add Stream.forEach

tests/lib/async/stream_controller_async_test.dart

 // Copyright (c) 2011, 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.
 
 // Test the basic StreamController and StreamController.singleSubscription.
 library stream_controller_async_test;
 
 import "package:expect/expect.dart";
 import 'dart:async';
 import 'dart:isolate';
@@ skipping 107 matching lines @@
     sink.add(3); // -"-
     sink.add(4); // -"-
     sink.add(5); // seen by stream 10
     sink.close();
   });
 }
 
 testExtraMethods() {
   Events sentEvents = new Events()..add(7)..add(9)..add(13)..add(87)..close();
 
+  test("forEach", () {
+    StreamController c = new StreamController();
+    Events actualEvents = new Events();
+    Future f = c.stream.forEach(actualEvents.add);
+    f.then(expectAsync1((_) {
+      actualEvents.close();
+      Expect.listEquals(sentEvents.events, actualEvents.events);
+    }));
+    sentEvents.replay(c);
+  });
+
+  test("forEachError", () {
+    Events sentEvents = new Events()..add(7)..error("bad")..add(87)..close();
+    StreamController c = new StreamController();
+    Events actualEvents = new Events();
+    Future f = c.stream.forEach(actualEvents.add);
+    f.catchError(expectAsync1((error) {
+      Expect.equals("bad", error);
+      Expect.listEquals((new Events()..add(7)).events, actualEvents.events);
+    }));
+    sentEvents.replay(c);
+  });
+
+  test("forEachError2", () {
+    Events sentEvents = new Events()..add(7)..add(9)..add(87)..close();
+    StreamController c = new StreamController();
+    Events actualEvents = new Events();
+    Future f = c.stream.forEach((x) {
+      if (x == 9) throw "bad";
+      actualEvents.add(x);
+    });
+    f.catchError(expectAsync1((error) {
+      Expect.equals("bad", error);
+      Expect.listEquals((new Events()..add(7)).events, actualEvents.events);
+    }));
+    sentEvents.replay(c);
+  });
+
   test("firstWhere", () {
     StreamController c = new StreamController();
     Future f = c.stream.firstWhere((x) => (x % 3) == 0);
     f.then(expectAsync1((v) { Expect.equals(9, v); }));
     sentEvents.replay(c);
   });
 
   test("firstWhere 2", () {
     StreamController c = new StreamController();
     Future f = c.stream.firstWhere((x) => (x % 4) == 0);
@@ skipping 228 matching lines @@
       Expect.listEquals(expectedEvents.events, actualEvents.events);
     }));
     actualEvents.resume();
   });
 }
 
 class TestError { const TestError(); }
 
 testRethrow() {
   TestError error = const TestError();
-  
+
 
   testStream(name, streamValueTransform) {
     test("rethrow-$name-value", () {
       StreamController c = new StreamController();
       Stream s = streamValueTransform(c.stream, (v) { throw error; });
       s.listen((_) { Expect.fail("unexpected value"); }, onError: expectAsync1(
           (e) { Expect.identical(error, e); }));
       c.add(null);
       c.close();
     });
@@ skipping 22 matching lines @@
       c.close();
     });
   }
 
   testStream("where", (s, act) => s.where(act));
   testStream("map", (s, act) => s.map(act));
   testStream("expand", (s, act) => s.expand(act));
   testStream("where", (s, act) => s.where(act));
   testStreamError("handleError", (s, act) => s.handleError(act));
   testStreamError("handleTest", (s, act) => s.handleError((v) {}, test: act));
+  testFuture("forEach", (s, act) => s.forEach(act));
   testFuture("every", (s, act) => s.every(act));
   testFuture("any", (s, act) => s.any(act));
   testFuture("reduce", (s, act) => s.reduce((a,b) => act(b)));
   testFuture("fold", (s, act) => s.fold(0, (a,b) => act(b)));
 }
 
 main() {
   testController();
   testSingleController();
   testExtraMethods();
   testPause();
   testRethrow();
 }