Add StreamController.multiplex constructor.

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 394 matching lines @@
   testStream("where", (s, act) => s.where(act));
   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)));
 }
 
+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();
 }