Make StreamController be a StreamSink, not just an EventSink.

tests/lib/async/event_helper.dart

 // Copyright (c) 2012, 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.
 
 library event_helper;
 
 import 'dart:async';
 
 abstract class Event {
   void replay(EventSink sink);
@@ skipping 43 matching lines @@
   int get hashCode => 42;
 
   bool operator==(Object other) => other is DoneEvent;
 
   String toString() => "DoneEvent";
 }
 
 /** Collector of events. */
 class Events implements EventSink {
   final List<Event> events = [];
+  bool trace = false;
 
   Events();
+
   Events.fromIterable(Iterable iterable) {
     for (var value in iterable) add(value);
     close();
   }
 
   /** Capture events from a stream into a new [Events] object. */
   factory Events.capture(Stream stream,
                          { bool cancelOnError: false }) = CaptureEvents;
 
   // EventSink interface.
   void add(var value) {
+    if (trace) print("Events#$hashCode: add($value)");
     events.add(new DataEvent(value));
   }
 
   void addError(error) {
+    if (trace) print("Events#$hashCode: addError($error)");
     events.add(new ErrorEvent(error));
   }
 
   void close() {
+    if (trace) print("Events#$hashCode: close()");
     events.add(const DoneEvent());
   }
 
   // Error helper for creating errors manually..
   void error(var value) { addError(value); }
 
   /** Replay the captured events on a sink. */
   void replay(EventSink sink) {
     for (int i = 0; i < events.length; i++) {
       events[i].replay(sink);
@@ skipping 55 matching lines @@
     super.addError(error);
     if (cancelOnError) onDoneSignal.complete(null);
   }
 
   void close() {
     super.close();
     if (onDoneSignal != null) onDoneSignal.complete(null);
   }
 
   void pause([Future resumeSignal]) {
+    if (trace) print("Events#$hashCode: pause");
     subscription.pause(resumeSignal);
   }
 
   void resume() {
+    if (trace) print("Events#$hashCode: resume");
     subscription.resume();
   }
 
   void onDone(void action()) {
+    if (trace) print("Events#$hashCode: onDone");
     onDoneSignal.future.whenComplete(action);
   }
 }