Reapply "Active stream subscriptions".

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';
@@ skipping 35 matching lines @@
   });
 
   test("Single-subscription StreamController.fold throws", () {
     StreamController c = new StreamController();
     Stream stream = c.stream;
     stream.fold(0, (a,b) { throw "Fnyf!"; })
             .catchError(expectAsync1((e) { Expect.equals("Fnyf!", e); }));
     c.add(42);
   });
 
-  test("Single-subscription StreamController subscription changes", () {
-    StreamController c = new StreamController();
-    EventSink sink = c.sink;
-    Stream stream = c.stream;
-    int counter = 0;
-    var subscription;
-    subscription = stream.listen((data) {
-      counter += data;
-      Expect.throws(() => stream.listen(null), (e) => e is StateError);
-      subscription.cancel();
-      stream.listen((data) {
-        counter += data * 10;
-      },
-      onDone: expectAsync0(() {
-        Expect.equals(1 + 20, counter);
-      }));
-    });
-    sink.add(1);
-    sink.add(2);
-    sink.close();
-  });
-
   test("Single-subscription StreamController events are buffered when"
        " there is no subscriber",
        () {
     StreamController c = new StreamController();
     EventSink sink = c.sink;
     Stream stream = c.stream;
     int counter = 0;
     sink.add(1);
     sink.add(2);
     sink.close();
     stream.listen(
       (data) {
         counter += data;
       },
       onDone: expectAsync0(() {
         Expect.equals(3, counter);
       }));
   });
-
-  // Test subscription changes while firing.
-  test("Single-subscription StreamController subscription changes while firing",
-       () {
-    StreamController c = new StreamController();
-    EventSink sink = c.sink;
-    Stream stream = c.stream;
-    int counter = 0;
-    var subscription = stream.listen(null);
-    subscription.onData(expectAsync1((data) {
-      counter += data;
-      subscription.cancel();
-      stream.listen((data) {
-        counter += 10 * data;
-      },
-      onDone: expectAsync0(() {
-        Expect.equals(1 + 20 + 30 + 40 + 50, counter);
-      }));
-      Expect.throws(() => stream.listen(null), (e) => e is StateError);
-    }));
-    sink.add(1); // seen by stream 1
-    sink.add(2); // seen by stream 10 and 100
-    sink.add(3); // -"-
-    sink.add(4); // -"-
-    sink.add(5); // seen by stream 10
-    sink.close();
-  });
 }
 
 testExtraMethods() {
   Events sentEvents = new Events()..add(7)..add(9)..add(13)..add(87)..close();
 
   test("forEach", () {
     StreamController c = new StreamController();
     Events actualEvents = new Events();
     Future f = c.stream.forEach(actualEvents.add);
     f.then(expectAsync1((_) {
@@ skipping 338 matching lines @@
   testStreamError("handleError", (s, act) => s.handleError(act));
   testStreamError("handleTest", (s, act) => s.handleError((v) {}, test: act));
   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 testMultiplex() {
+  test("multiplex-basic", () {
+    StreamController<int> c = new StreamController.multiplex(
+      onListen: expectAsync0(() {}),
+      onCancel: expectAsync0(() {})
+    );
+    Stream<int> s = c.stream;
+    s.listen(expectAsync1((x) { expect(x, equals(42)); }));
+    c.add(42);
+    c.close();
+  });
+
+  test("multiplex-listen-twice", () {
+    StreamController<int> c = new StreamController.multiplex(
+      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();
+  });
+
+  test("multiplex-listen-twice-non-overlap", () {
+    StreamController<int> c = new StreamController.multiplex(
+      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();
+  });
+
+  test("multiplex-individual-pause", () {
+    StreamController<int> c = new StreamController.multiplex(
+      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();
+  });
+}
+
 main() {
   testController();
   testSingleController();
   testExtraMethods();
   testPause();
   testRethrow();
+  testMultiplex();
 }