Fix flaky stream_controller_async_test.

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 44 matching lines @@
 
   bool operator==(Object other) => other is DoneEvent;
 
   String toString() => "DoneEvent";
 }
 
 /** Collector of events. */
 class Events implements EventSink {
   final List<Event> events = [];
   bool trace = false;
+  Completer onDoneSignal = new Completer();
 
   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, [StackTrace stackTrace]) {
     if (trace) print("Events#$hashCode: addError($error)");
     events.add(new ErrorEvent(error));
   }
 
   void close() {
     if (trace) print("Events#$hashCode: close()");
     events.add(const DoneEvent());
+    onDoneSignal.complete();
   }
 
-  // Error helper for creating errors manually..
+  /**
+   * Error shorthand, for writing events manually.
+   */
   void error(var value, [StackTrace stackTrace]) {
     addError(value, stackTrace);
   }
 
   /** Replay the captured events on a sink. */
   void replay(EventSink sink) {
     for (int i = 0; i < events.length; i++) {
       events[i].replay(sink);
     }
   }
@@ skipping 21 matching lines @@
     throw new StateError("Not capturing events.");
   }
 
   /** Resumes after a call to [pause]. */
   void resume() {
     throw new StateError("Not capturing events.");
   }
 
   /**
    * Sets an action to be called when this [Events] receives a 'done' event.
+   *
+   * The action will also be called if capturing events from a stream with
+   * `cancelOnError` set to true and receiving an error.
    */
   void onDone(void action()) {
-     throw new StateError("Not capturing events.");
+    onDoneSignal.future.whenComplete(action);
   }
 }
 
 class CaptureEvents extends Events {
   StreamSubscription subscription;
-  Completer onDoneSignal;
   bool cancelOnError = false;
 
   CaptureEvents(Stream stream,
-                { bool cancelOnError: false })
-      : onDoneSignal = new Completer() {
+                { bool cancelOnError: false }) {
     this.cancelOnError = cancelOnError;
     subscription = stream.listen(add,
                                  onError: addError,
                                  onDone: close,
                                  cancelOnError: cancelOnError);
   }
 
   void addError(error) {
     super.addError(error);
-    if (cancelOnError) onDoneSignal.complete(null);
-  }
-
-  void close() {
-    super.close();
-    if (onDoneSignal != null) onDoneSignal.complete(null);
+    if (cancelOnError) {
+      onDoneSignal.complete();
+    }
   }
 
   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);
+    super.onDone(action);
   }
 }