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

tests/lib/async/stream_event_transform_test.dart

Index: tests/lib/async/stream_event_transform_test.dart
diff --git a/tests/lib/async/stream_event_transform_test.dart b/tests/lib/async/stream_event_transform_test.dart
new file mode 100644
index 0000000000000000000000000000000000000000..3acd3f0b15e9d888202138e938c143db1eef8799
--- /dev/null
+++ b/tests/lib/async/stream_event_transform_test.dart
@@ -0,0 +1,77 @@
+// 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.
+
+import 'dart:async';
+import '../../../pkg/unittest/lib/unittest.dart';
+import 'event_helper.dart';
+
+void handleData(int data, StreamSink<int> sink) {
+  sink.signalError(new AsyncError("$data"));
+  sink.add(data + 1);
+}
+void handleError(AsyncError e, StreamSink<int> sink) {
+  String value = e.error;
+  int data = int.parse(value);
+  sink.add(data);
+  sink.signalError(new AsyncError("${data + 1}"));
+}
+
+void handleDone(StreamSink<int> sink) {
+  sink.add(99);
+  sink.close();
+}
+
+class EventTransformer extends StreamEventTransformer<int,int> {
+  void handleData(int data, StreamSink<int> sink) {
+    sink.signalError(new AsyncError("$data"));
+    sink.add(data + 1);
+  }
+  void handleError(AsyncError e, StreamSink<int> sink) {
+    String value = e.error;
+    int data = int.parse(value);
+    sink.add(data);
+    sink.signalError(new AsyncError("${data + 1}"));
+  }
+
+  void handleDone(StreamSink<int> sink) {
+    sink.add(99);
+    sink.close();
+  }
+}
+
+main() {
+  {
+    StreamController c = new StreamController();
+    Events expected = new Events()..error("0")..add(1)
+                                  ..error("1")..add(2)
+                                  ..add(3)..error("4")
+                                  ..add(99)..close();
+    Events input = new Events()..add(0)..add(1)..error("3")..close();
+    Events actual = new Events.capture(
+        c.stream.transformEvents(new EventTransformer()));
+    actual.onDone(() {
+      Expect.listEquals(expected.events, actual.events);
+    });
+    input.replay(c);
+  }
+
+  {
+    StreamController c = new StreamController();
+    Events expected = new Events()..error("0")..add(1)
+                                  ..error("1")..add(2)
+                                  ..add(3)..error("4")
+                                  ..add(99)..close();
+    Events input = new Events()..add(0)..add(1)..error("3")..close();
+    Events actual = new Events.capture(
+        c.stream.transformEvents(new StreamEventTransformer.from(
+            handleData: handleData,
+            handleError: handleError,
+            handleDone: handleDone
+        )));
+    actual.onDone(() {
+      Expect.listEquals(expected.events, actual.events);
+    });
+    input.replay(c);
+  }
+}