Implement asBroadcast using a _MultiplexStreamController.

tests/lib/async/stream_controller_async_test.dart

Index: tests/lib/async/stream_controller_async_test.dart
diff --git a/tests/lib/async/stream_controller_async_test.dart b/tests/lib/async/stream_controller_async_test.dart
index 1ed514372a13dead90d79a636c1dc2b00056bf5c..65e658f32b259b8cf2f236ae969566fe3061d698 100644
--- a/tests/lib/async/stream_controller_async_test.dart
+++ b/tests/lib/async/stream_controller_async_test.dart
@@ -482,6 +482,25 @@ void testBroadcastController() {
     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() {