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

tests/lib/async/stream_controller_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_test;
 
 import 'dart:async';
 import 'event_helper.dart';
 
@@ skipping 222 matching lines @@
   var list = <int>[];
   c.stream.pipeInto(new CollectionSink<int>(list))
    .whenComplete(() { Expect.listEquals(<int>[1,2,9,3,9], list); });
   c.add(1);
   c.add(2);
   c.add(9);
   c.add(3);
   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();
   expectedEvents =
       new Events()..error("a")..add(42)..error("b")..add("foo")..close();
   actualEvents = new Events.capture(c.stream.transform(
       new StreamTransformer.from(
           onData: (v, s) { s.signalError(new AsyncError(v)); },
           onError: (e, s) { s.add(e.error); },
           onDone: (s) {
@@ skipping 101 matching lines @@
   actualEvents = new Events.capture(c.stream.distinct((a, b) => a < b));
   sentEvents.replay(c);
   Expect.listEquals(expectedEvents.events, actualEvents.events);
 }
 
 main() {
   testMultiController();
   testSingleController();
   testExtraMethods();
 }