Make new StreamController be async by default.

tests/lib/async/stream_controller_async_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_async_test;
 
 import "package:expect/expect.dart";
 import 'dart:async';
 import 'dart:isolate';
 import '../../../pkg/unittest/lib/unittest.dart';
 import 'event_helper.dart';
+import 'stream_state_helper.dart';
 
 testController() {
   // Test fold
   test("StreamController.fold", () {
     StreamController c = new StreamController();
     Stream stream = c.stream.asBroadcastStream();
     stream.fold(0, (a,b) => a + b)
      .then(expectAsync1((int v) {
         Expect.equals(42, v);
     }));
@@ skipping 236 matching lines @@
     Future f = c.stream.drain();
     f.catchError(expectAsync1((error) { Expect.equals("error", error); }));
     Events errorEvents = new Events()..error("error")..error("error2")..close();
     errorEvents.replay(c);
   });
 
 }
 
 testPause() {
   test("pause event-unpause", () {
-    StreamController c = new StreamController();
-    Events actualEvents = new Events.capture(c.stream);
-    Events expectedEvents = new Events();
-    expectedEvents.add(42);
-    c.add(42);
-    Expect.listEquals(expectedEvents.events, actualEvents.events);
+    StreamProtocolTest test = new StreamProtocolTest();
     Completer completer = new Completer();
-    actualEvents.pause(completer.future);
-    c..add(43)..add(44)..close();
-    Expect.listEquals(expectedEvents.events, actualEvents.events);
-    completer.complete();
-    expectedEvents..add(43)..add(44)..close();
-    actualEvents.onDone(expectAsync0(() {
-      Expect.listEquals(expectedEvents.events, actualEvents.events);
-    }));
+    test..expectListen()
+        ..expectData(42, () { test.pause(completer.future); })
+        ..expectPause(() {
+            completer.complete(null);
+          })
+        ..expectData(43)
+        ..expectData(44)
+        ..expectDone()
+        ..expectCancel();
+    test.listen();
+    test.add(42);
+    test.add(43);
+    test.add(44);
+    test.close();
   });
 
   test("pause twice event-unpause", () {
-    StreamController c = new StreamController();
-    Events actualEvents = new Events.capture(c.stream);
-    Events expectedEvents = new Events();
-    expectedEvents.add(42);
-    c.add(42);
-    Expect.listEquals(expectedEvents.events, actualEvents.events);
+    StreamProtocolTest test = new StreamProtocolTest();
     Completer completer = new Completer();
     Completer completer2 = new Completer();
-    actualEvents.pause(completer.future);
-    actualEvents.pause(completer2.future);
-    c..add(43)..add(44)..close();
-    Expect.listEquals(expectedEvents.events, actualEvents.events);
-    completer.complete();
-    Expect.listEquals(expectedEvents.events, actualEvents.events);
-    completer2.complete();
-    expectedEvents..add(43)..add(44)..close();
-    actualEvents.onDone(expectAsync0((){
-      Expect.listEquals(expectedEvents.events, actualEvents.events);
-    }));
+    test..expectListen()
+        ..expectData(42, () {
+            test.pause(completer.future);
+            test.pause(completer2.future);
+          })
+        ..expectPause(() {
+            completer.future.then(completer2.complete);
+            completer.complete(null);
+          })
+        ..expectData(43)
+        ..expectData(44)
+        ..expectDone()
+        ..expectCancel();
+    test..listen()
+        ..add(42)
+        ..add(43)
+        ..add(44)
+        ..close();
   });
 
   test("pause twice direct-unpause", () {
-    StreamController c = new StreamController();
-    Events actualEvents = new Events.capture(c.stream);
-    Events expectedEvents = new Events();
-    expectedEvents.add(42);
-    c.add(42);
-    Expect.listEquals(expectedEvents.events, actualEvents.events);
-    actualEvents.pause();
-    actualEvents.pause();
-    c.add(43);
-    c.add(44);
-    c.close();
-    Expect.listEquals(expectedEvents.events, actualEvents.events);
-    actualEvents.resume();
-    Expect.listEquals(expectedEvents.events, actualEvents.events);
-    expectedEvents..add(43)..add(44)..close();
-    actualEvents.onDone(expectAsync0(() {
-      Expect.listEquals(expectedEvents.events, actualEvents.events);
-    }));
-    actualEvents.resume();
+    StreamProtocolTest test = new StreamProtocolTest();
+    test..expectListen()
+        ..expectData(42, () {
+            test.pause();
+            test.pause();
+          })
+        ..expectPause(() {
+            test.resume();
+            test.resume();
+          })
+        ..expectData(43)
+        ..expectData(44)
+        ..expectDone()
+        ..expectCancel();
+    test..listen()
+        ..add(42)
+        ..add(43)
+        ..add(44)
+        ..close();
   });
 
   test("pause twice direct-event-unpause", () {
-    StreamController c = new StreamController();
-    Events actualEvents = new Events.capture(c.stream);
-    Events expectedEvents = new Events();
-    expectedEvents.add(42);
-    c.add(42);
-    Expect.listEquals(expectedEvents.events, actualEvents.events);
+    StreamProtocolTest test = new StreamProtocolTest();
     Completer completer = new Completer();
-    actualEvents.pause(completer.future);
-    actualEvents.pause();
-    c.add(43);
-    c.add(44);
-    c.close();
-    Expect.listEquals(expectedEvents.events, actualEvents.events);
-    actualEvents.resume();
-    Expect.listEquals(expectedEvents.events, actualEvents.events);
-    expectedEvents..add(43)..add(44)..close();
-    actualEvents.onDone(expectAsync0(() {
-      Expect.listEquals(expectedEvents.events, actualEvents.events);
-    }));
-    completer.complete();
-  });
-
-  test("pause twice direct-unpause", () {
-    StreamController c = new StreamController();
-    Events actualEvents = new Events.capture(c.stream);
-    Events expectedEvents = new Events();
-    expectedEvents.add(42);
-    c.add(42);
-    Expect.listEquals(expectedEvents.events, actualEvents.events);
-    Completer completer = new Completer();
-    actualEvents.pause(completer.future);
-    actualEvents.pause();
-    c.add(43);
-    c.add(44);
-    c.close();
-    Expect.listEquals(expectedEvents.events, actualEvents.events);
-    completer.complete();
-    Expect.listEquals(expectedEvents.events, actualEvents.events);
-    expectedEvents..add(43)..add(44)..close();
-    actualEvents.onDone(expectAsync0(() {
-      Expect.listEquals(expectedEvents.events, actualEvents.events);
-    }));
-    actualEvents.resume();
+    test..expectListen()
+        ..expectData(42, () {
+            test.pause();
+            test.pause(completer.future);
+            test.add(43);
+            test.add(44);
+            test.close();
+          })
+        ..expectPause(() {
+            completer.future.then((v) => test.resume());
+            completer.complete(null);
+          })
+        ..expectData(43)
+        ..expectData(44)
+        ..expectDone()
+        ..expectCancel();
+    test..listen()
+        ..add(42);
   });
 }
 
 class TestError { const TestError(); }
 
 testRethrow() {
   TestError error = const TestError();
 
 
   testStream(name, streamValueTransform) {
@@ skipping 40 matching lines @@
   testFuture("forEach", (s, act) => s.forEach(act));
   testFuture("every", (s, act) => s.every(act));
   testFuture("any", (s, act) => s.any(act));
   testFuture("reduce", (s, act) => s.reduce((a,b) => act(b)));
   testFuture("fold", (s, act) => s.fold(0, (a,b) => act(b)));
   testFuture("drain", (s, act) => s.drain().then(act));
 }
 
 void testBroadcastController() {
   test("broadcast-controller-basic", () {
-    StreamController<int> c = new StreamController.broadcast(
-      onListen: expectAsync0(() {}),
-      onCancel: expectAsync0(() {})
-    );
-    Stream<int> s = c.stream;
-    s.listen(expectAsync1((x) { expect(x, equals(42)); }));
-    c.add(42);
-    c.close();
+    StreamProtocolTest test = new StreamProtocolTest.broadcast();
+    test..expectListen()
+        ..expectData(42)
+        ..expectDone()
+        ..expectCancel(test.terminate);
+    test..listen()
+        ..add(42)
+        ..close();
   });
 
   test("broadcast-controller-listen-twice", () {
-    StreamController<int> c = new StreamController.broadcast(
-      onListen: expectAsync0(() {}),
-      onCancel: expectAsync0(() {})
-    );
-    c.stream.listen(expectAsync1((x) { expect(x, equals(42)); }, count: 2));
-    c.add(42);
-    c.stream.listen(expectAsync1((x) { expect(x, equals(42)); }));
-    c.add(42);
-    c.close();
+    StreamProtocolTest test = new StreamProtocolTest.broadcast();
+    test..expectListen()
+        ..expectData(42, () {
+            test.listen();
+            test.add(37);
+            test.close();
+          })
+      // Order is not guaranteed between subscriptions if not sync.
+        ..expectData(37)
+        ..expectData(37)
+        ..expectDone()
+        ..expectDone()
+        ..expectCancel(test.terminate);
+    test.listen();
+    test.add(42);
   });
 
   test("broadcast-controller-listen-twice-non-overlap", () {
-    StreamController<int> c = new StreamController.broadcast(
-      onListen: expectAsync0(() {}, count: 2),
-      onCancel: expectAsync0(() {}, count: 2)
-    );
-    var sub = c.stream.listen(expectAsync1((x) { expect(x, equals(42)); }));
-    c.add(42);
-    sub.cancel();
-    c.stream.listen(expectAsync1((x) { expect(x, equals(42)); }));
-    c.add(42);
-    c.close();
+    StreamProtocolTest test = new StreamProtocolTest.broadcast();
+    test
+    ..expectListen(() {
+      test.add(42);
+    })
+    ..expectData(42, () {
+      test.cancel();
+    })
+    ..expectCancel(() {
+      test.listen();
+    })..expectListen(() {
+      test.add(37);
+    })
+    ..expectData(37, () {
+      test.close();
+    })
+    ..expectDone()
+    ..expectCancel(test.terminate);
+    test.listen();
   });
 
   test("broadcast-controller-individual-pause", () {
-    StreamController<int> c = new StreamController.broadcast(
-      onListen: expectAsync0(() {}),
-      onCancel: expectAsync0(() {})
-    );
-    var sub1 = c.stream.listen(expectAsync1((x) { expect(x, equals(42)); }));
-    var sub2 = c.stream.listen(expectAsync1((x) { expect(x, equals(42)); },
-                                            count: 3));
-    c.add(42);
-    sub1.pause();
-    c.add(42);
-    sub1.cancel();
-    var sub3 = c.stream.listen(expectAsync1((x) { expect(x, equals(42)); }));
-    c.add(42);
-    c.close();
+    StreamProtocolTest test = new StreamProtocolTest.broadcast();
+    test.trace = true;
+    var sub1;
+    test..expectListen()
+        ..expectData(42)
+        ..expectData(42, () { sub1.pause(); })
+        ..expectData(43, () {
+      sub1.cancel();
+      test.listen();
+      test.add(44);
+      test.expectData(44);
+      test.expectData(44, test.terminate);
+    });
+    sub1 = test.listen();
+    test.listen();
+    test.add(42);
+    test.add(43);
   });
 
   test("broadcast-controller-add-in-callback", () {
-    StreamController<int> c;
-    c = new StreamController(
-      onListen: expectAsync0(() {}),
-      onCancel: expectAsync0(() {
-        c.add(42);
-      })
-    );
-    var sub;
-    sub = c.stream.asBroadcastStream().listen(expectAsync1((v) {
-      Expect.equals(37, v);
-      c.add(21);
+    StreamProtocolTest test = new StreamProtocolTest.broadcast();
+    test.expectListen();
+    var sub = test.listen();
+    test.add(42);
+    sub.expectData(42, () {
+      test.add(87);
       sub.cancel();
-    }));
-    c.add(37);  // Triggers listener, which adds 21 and removes itself.
-    // Removing listener triggers onCancel which adds another 42.
-    // Both 21 and 42 are lost because there are no listeners.
+    });
+    test.expectCancel(() {
+      test.add(37);
+      test.terminate();
+    });
   });
 }
 
 main() {
   testController();
   testSingleController();
   testExtraMethods();
   testPause();
   testRethrow();
   testBroadcastController();
 }