Add public-facing method and class that allows intercepting stream events.

tests/lib/async/stream_controller_test.dart

Index: tests/lib/async/stream_controller_test.dart
diff --git a/tests/lib/async/stream_controller_test.dart b/tests/lib/async/stream_controller_test.dart
index 9dd7509fad0e094b28a5f748ebd33fa245a4d80c..57b99e3699b5395100e8584cbcf627adbe2c6e3c 100644
--- a/tests/lib/async/stream_controller_test.dart
+++ b/tests/lib/async/stream_controller_test.dart
@@ -240,6 +240,42 @@ testSingleController() {
   c.add(9);
   c.close();
 
+  // test contains.
+  {
+    c = new StreamController();
+    // Error after match is not important.
+    sentEvents = new Events()..add("a")..add("x")..error("FAIL")..close();
+    Future<bool> contains = c.stream.contains("x");
+    contains.then((var c) {
+      Expect.isTrue(c);
+    });
+    sentEvents.replay(c);
+  }
+
+  {
+    c = new StreamController();
+    // Not matching is ok.
+    sentEvents = new Events()..add("a")..add("x")..add("b")..close();
+    Future<bool> contains = c.stream.contains("y");
+    contains.then((var c) {
+      Expect.isFalse(c);
+    });
+    sentEvents.replay(c);
+  }
+
+  {
+    c = new StreamController();
+    // Error before match makes future err.
+    sentEvents = new Events()..add("a")..error("FAIL")..add("b")..close();
+    Future<bool> contains = c.stream.contains("b");
+    contains.then((var c) {
+      Expect.fail("no value expected");
+    }).catchError((AsyncError e) {
+      Expect.equals("FAIL", e.error);
+    });
+    sentEvents.replay(c);
+  }
+
   // Test transform.
   c = new StreamController();
   sentEvents = new Events()..add("a")..error(42)..add("b")..close();