Roll Observatory deps (charted -> ^0.3.0)

packages/async/test/subscription_stream_test.dart

 // Copyright (c) 2015, 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 "package:async/async.dart" show SubscriptionStream;
 import "package:test/test.dart";
 
 import "utils.dart";
@@ skipping 55 matching lines @@
     expect(controller.isPaused, isFalse);
 
     expect(await subscription.cancel(), 42);
     expect(controller.hasListener, isFalse);
   });
 
   group("cancelOnError source:", () {
     for (var sourceCancels in [false, true]) {
       group("${sourceCancels ? "yes" : "no"}:", () {
         var subscriptionStream;
+        var onCancel;  // Completes if source stream is canceled before done.
         setUp(() {
-          var source = createErrorStream();
+          var cancelCompleter = new Completer();
+          var source = createErrorStream(cancelCompleter);
+          onCancel = cancelCompleter.future;
           var sourceSubscription = source.listen(null,
                                                  cancelOnError: sourceCancels);
           subscriptionStream = new SubscriptionStream<int>(sourceSubscription);
         });
 
         test("- subscriptionStream: no", () async {
           var done = new Completer();
           var events = [];
           subscriptionStream.listen(events.add,
                                     onError: events.add,
                                     onDone: done.complete,
                                     cancelOnError: false);
           var expected = [1, 2, "To err is divine!"];
           if (sourceCancels) {
-            var timeout = done.future.timeout(const Duration(milliseconds: 5),
-                                              onTimeout: () => true);
-            expect(await timeout, true);
+            await onCancel;
+            // And [done] won't complete at all.
+            bool isDone = false;
+            done.future.then((_) { isDone = true; });
+            await new Future.delayed(const Duration(milliseconds: 5));
+            expect(isDone, false);
           } else {
             expected.add(4);
             await done.future;
           }
           expect(events, expected);
         });
 
         test("- subscriptionStream: yes", () async {
           var completer = new Completer();
           var events = [];
@@ skipping 43 matching lines @@
 Stream<int> createStream() async* {
   yield 1;
   await flushMicrotasks();
   yield 2;
   await flushMicrotasks();
   yield 3;
   await flushMicrotasks();
   yield 4;
 }
 
-Stream<int> createErrorStream() {
-  StreamController controller = new StreamController<int>();
-  () async {
-    controller.add(1);
+Stream<int> createErrorStream([Completer onCancel]) async* {
+  bool canceled = true;
+  try {
+    yield 1;
     await flushMicrotasks();
-    controller.add(2);
+    yield 2;
     await flushMicrotasks();
-    controller.addError("To err is divine!");
+    yield* new Future.error("To err is divine!").asStream();
     await flushMicrotasks();
-    controller.add(4);
+    yield 4;
     await flushMicrotasks();
-    controller.close();
-  }();
-  return controller.stream;
+    canceled = false;
+  } finally {
+    // Completes before the "done", but should be after all events.
+    if (canceled && onCancel != null) {
+      await flushMicrotasks();
+      onCancel.complete();
+    }
+  }
 }
 
 Stream<int> createLongStream() async* {
   for (int i = 0; i < 200; i++) yield i;
 }