Change singleSubscription/multiSubscription to normal/broadcast.

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 'dart:async';
 import 'dart:isolate';
 import '../../../pkg/unittest/lib/unittest.dart';
 import 'event_helper.dart';
 
 testController() {
   // Test reduce
   test("StreamController.reduce", () {
-    StreamController c = new StreamController.multiSubscription();
-    c.reduce(0, (a,b) => a + b)
+    StreamController c = new StreamController.broadcast();
+    Stream stream = c.stream;
+    stream.reduce(0, (a,b) => a + b)
      .then(expectAsync1((int v) {
         Expect.equals(42, v);
     }));
     c.add(10);
     c.add(32);
     c.close();
   });
 
   test("StreamController.reduce throws", () {
-    StreamController c = new StreamController.multiSubscription();
-    c.reduce(0, (a,b) { throw "Fnyf!"; })
+    StreamController c = new StreamController.broadcast();
+    Stream stream = c.stream;
+    stream.reduce(0, (a,b) { throw "Fnyf!"; })
      .catchError(expectAsync1((e) { Expect.equals("Fnyf!", e.error); }));
     c.add(42);
   });
 
   test("StreamController.pipeInto", () {
-    StreamController c = new StreamController.multiSubscription();
+    StreamController c = new StreamController.broadcast();
     var list = <int>[];
-    c.pipeInto(new CollectionSink<int>(list))
+    Stream stream = c.stream;
+    stream.pipeInto(new CollectionSink<int>(list))
      .whenComplete(expectAsync0(() {
         Expect.listEquals(<int>[1,2,9,3,9], list);
       }));
     c.add(1);
     c.add(2);
     c.add(9);
     c.add(3);
     c.add(9);
     c.close();
   });
 }
 
 testSingleController() {
   test("Single-subscription StreamController.reduce", () {
     StreamController c = new StreamController();
-    c.reduce(0, (a,b) => a + b)
+    Stream stream = c.stream;
+    stream.reduce(0, (a,b) => a + b)
     .then(expectAsync1((int v) { Expect.equals(42, v); }));
     c.add(10);
     c.add(32);
     c.close();
   });
 
   test("Single-subscription StreamController.reduce throws", () {
     StreamController c = new StreamController();
-    c.reduce(0, (a,b) { throw "Fnyf!"; })
+    Stream stream = c.stream;
+    stream.reduce(0, (a,b) { throw "Fnyf!"; })
             .catchError(expectAsync1((e) { Expect.equals("Fnyf!", e.error); }));
     c.add(42);
   });
 
   test("Single-subscription StreamController.pipeInto", () {
     StreamController c = new StreamController();
     var list = <int>[];
-    c.pipeInto(new CollectionSink<int>(list))
+    Stream stream = c.stream;
+    stream.pipeInto(new CollectionSink<int>(list))
      .whenComplete(expectAsync0(() {
         Expect.listEquals(<int>[1,2,9,3,9], list);
       }));
     c.add(1);
     c.add(2);
     c.add(9);
     c.add(3);
     c.add(9);
     c.close();
   });
@@ skipping 65 matching lines @@
     sink.add(5); // seen by stream 10
     sink.close();
   });
 }
 
 testExtraMethods() {
   Events sentEvents = new Events()..add(7)..add(9)..add(13)..add(87)..close();
 
   test("firstMatching", () {
     StreamController c = new StreamController();
-    Future f = c.firstMatching((x) => (x % 3) == 0);
+    Future f = c.stream.firstMatching((x) => (x % 3) == 0);
     f.then(expectAsync1((v) { Expect.equals(9, v); }));
     sentEvents.replay(c);
   });
 
   test("firstMatching 2", () {
     StreamController c = new StreamController();
-    Future f = c.firstMatching((x) => (x % 4) == 0);
+    Future f = c.stream.firstMatching((x) => (x % 4) == 0);
     f.catchError(expectAsync1((e) {}));
     sentEvents.replay(c);
   });
 
   test("firstMatching 3", () {
     StreamController c = new StreamController();
-    Future f = c.firstMatching((x) => (x % 4) == 0, defaultValue: () => 999);
+    Future f = c.stream.firstMatching((x) => (x % 4) == 0, defaultValue: () => 999);
     f.then(expectAsync1((v) { Expect.equals(999, v); }));
     sentEvents.replay(c);
   });
 
 
   test("lastMatching", () {
     StreamController c = new StreamController();
-    Future f = c.lastMatching((x) => (x % 3) == 0);
+    Future f = c.stream.lastMatching((x) => (x % 3) == 0);
     f.then(expectAsync1((v) { Expect.equals(87, v); }));
     sentEvents.replay(c);
   });
 
   test("lastMatching 2", () {
     StreamController c = new StreamController();
-    Future f = c.lastMatching((x) => (x % 4) == 0);
+    Future f = c.stream.lastMatching((x) => (x % 4) == 0);
     f.catchError(expectAsync1((e) {}));
     sentEvents.replay(c);
   });
 
   test("lastMatching 3", () {
     StreamController c = new StreamController();
-    Future f = c.lastMatching((x) => (x % 4) == 0, defaultValue: () => 999);
+    Future f = c.stream.lastMatching((x) => (x % 4) == 0, defaultValue: () => 999);
     f.then(expectAsync1((v) { Expect.equals(999, v); }));
     sentEvents.replay(c);
   });
 
   test("singleMatching", () {
     StreamController c = new StreamController();
-    Future f = c.singleMatching((x) => (x % 9) == 0);
+    Future f = c.stream.singleMatching((x) => (x % 9) == 0);
     f.then(expectAsync1((v) { Expect.equals(9, v); }));
     sentEvents.replay(c);
   });
 
   test("singleMatching 2", () {
     StreamController c = new StreamController();
-    Future f = c.singleMatching((x) => (x % 3) == 0);  // Matches both 9 and 87..
+    Future f = c.stream.singleMatching((x) => (x % 3) == 0);  // Matches both 9 and 87..
     f.catchError(expectAsync1((e) { Expect.isTrue(e.error is StateError); }));
     sentEvents.replay(c);
   });
 
   test("first", () {
     StreamController c = new StreamController();
-    Future f = c.first;
+    Future f = c.stream.first;
     f.then(expectAsync1((v) { Expect.equals(7, v);}));
     sentEvents.replay(c);
   });
 
   test("first empty", () {
     StreamController c = new StreamController();
-    Future f = c.first;
+    Future f = c.stream.first;
     f.catchError(expectAsync1((e) { Expect.isTrue(e.error is StateError); }));
     Events emptyEvents = new Events()..close();
     emptyEvents.replay(c);
   });
 
   test("first error", () {
     StreamController c = new StreamController();
-    Future f = c.first;
+    Future f = c.stream.first;
     f.catchError(expectAsync1((e) { Expect.equals("error", e.error); }));
     Events errorEvents = new Events()..error("error")..close();
     errorEvents.replay(c);
   });
 
   test("first error 2", () {
     StreamController c = new StreamController();
-    Future f = c.first;
+    Future f = c.stream.first;
     f.catchError(expectAsync1((e) { Expect.equals("error", e.error); }));
     Events errorEvents = new Events()..error("error")..error("error2")..close();
     errorEvents.replay(c);
   });
 
   test("last", () {
     StreamController c = new StreamController();
-    Future f = c.last;
+    Future f = c.stream.last;
     f.then(expectAsync1((v) { Expect.equals(87, v);}));
     sentEvents.replay(c);
   });
 
   test("last empty", () {
     StreamController c = new StreamController();
-    Future f = c.last;
+    Future f = c.stream.last;
     f.catchError(expectAsync1((e) { Expect.isTrue(e.error is StateError); }));
     Events emptyEvents = new Events()..close();
     emptyEvents.replay(c);
   });
 
   test("last error", () {
     StreamController c = new StreamController();
-    Future f = c.last;
+    Future f = c.stream.last;
     f.catchError(expectAsync1((e) { Expect.equals("error", e.error); }));
     Events errorEvents = new Events()..error("error")..close();
     errorEvents.replay(c);
   });
 
   test("last error 2", () {
     StreamController c = new StreamController();
-    Future f = c.last;
+    Future f = c.stream.last;
     f.catchError(expectAsync1((e) { Expect.equals("error", e.error); }));
     Events errorEvents = new Events()..error("error")..error("error2")..close();
     errorEvents.replay(c);
   });
 
   test("elementAt", () {
     StreamController c = new StreamController();
-    Future f = c.elementAt(2);
+    Future f = c.stream.elementAt(2);
     f.then(expectAsync1((v) { Expect.equals(13, v);}));
     sentEvents.replay(c);
   });
 
   test("elementAt 2", () {
     StreamController c = new StreamController();
-    Future f = c.elementAt(20);
+    Future f = c.stream.elementAt(20);
     f.catchError(expectAsync1((e) { Expect.isTrue(e.error is StateError); }));
     sentEvents.replay(c);
   });
 }
 
 testPause() {
   test("pause event-unpause", () {
     StreamController c = new StreamController();
-    Events actualEvents = new Events.capture(c);
+    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);
     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("pause twice event-unpause", () {
     StreamController c = new StreamController();
-    Events actualEvents = new Events.capture(c);
+    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();
     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("pause twice direct-unpause", () {
     StreamController c = new StreamController();
-    Events actualEvents = new Events.capture(c);
+    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();
   });
 
   test("pause twice direct-event-unpause", () {
     StreamController c = new StreamController();
-    Events actualEvents = new Events.capture(c);
+    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);
     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);
+    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();
@@ skipping 59 matching lines @@
   testFuture("reduce", (s, act) => s.reduce(0, (a,b) => act(b)));
 }
 
 main() {
   testController();
   testSingleController();
   testExtraMethods();
   testPause();
   testRethrow();
 }