Implement asBroadcast using a _MultiplexStreamController.

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 464 matching lines @@
     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();
   });
+
+  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);
+      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.
+  });
 }
 
 main() {
   testController();
   testSingleController();
   testExtraMethods();
   testPause();
   testRethrow();
   testBroadcastController();
 }