Make asBroadcastStream take two callbacks.

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';
 import '../../../pkg/unittest/lib/unittest.dart';
 import 'event_helper.dart';
 import 'stream_state_helper.dart';
 
+void cancelSub(StreamSubscription sub) { sub.cancel(); }
+
 testController() {
   // Test fold
   test("StreamController.fold", () {
     StreamController c = new StreamController();
-    Stream stream = c.stream.asBroadcastStream();
+    Stream stream = c.stream.asBroadcastStream(onCancel: cancelSub);
     stream.fold(0, (a,b) => a + b)
      .then(expectAsync1((int v) {
         Expect.equals(42, v);
     }));
     c.add(10);
     c.add(32);
     c.close();
   });
 
   test("StreamController.fold throws", () {
     StreamController c = new StreamController();
-    Stream stream = c.stream.asBroadcastStream();
+    Stream stream = c.stream.asBroadcastStream(onCancel: cancelSub);
     stream.fold(0, (a,b) { throw "Fnyf!"; })
      .catchError(expectAsync1((error) { Expect.equals("Fnyf!", error); }));
     c.add(42);
   });
 }
 
 testSingleController() {
   test("Single-subscription StreamController.fold", () {
     StreamController c = new StreamController();
     Stream stream = c.stream;
@@ skipping 218 matching lines @@
     Future f = c.stream.drain();
     f.catchError(expectAsync1((error) { Expect.equals("error", error); }));
     Events errorEvents = new Events()..error("error")..error("error2")..close();
     errorEvents.replay(c);
   });
 
 }
 
 testPause() {
   test("pause event-unpause", () {
+
     StreamProtocolTest test = new StreamProtocolTest();
     Completer completer = new Completer();
     test..expectListen()
         ..expectData(42, () { test.pause(completer.future); })
         ..expectPause(() {
             completer.complete(null);
           })
         ..expectData(43)
         ..expectData(44)
         ..expectDone()
-        ..expectCancel();
+        ..expectCancel(test.terminate);
     test.listen();
     test.add(42);
     test.add(43);
     test.add(44);
     test.close();
   });
 
   test("pause twice event-unpause", () {
     StreamProtocolTest test = new StreamProtocolTest();
     Completer completer = new Completer();
     Completer completer2 = new Completer();
     test..expectListen()
         ..expectData(42, () {
             test.pause(completer.future);
             test.pause(completer2.future);
           })
         ..expectPause(() {
             completer.future.then(completer2.complete);
             completer.complete(null);
           })
         ..expectData(43)
         ..expectData(44)
         ..expectDone()
-        ..expectCancel();
+        ..expectCancel(test.terminate);
     test..listen()
         ..add(42)
         ..add(43)
         ..add(44)
         ..close();
   });
 
   test("pause twice direct-unpause", () {
     StreamProtocolTest test = new StreamProtocolTest();
     test..expectListen()
         ..expectData(42, () {
             test.pause();
             test.pause();
           })
         ..expectPause(() {
             test.resume();
             test.resume();
           })
         ..expectData(43)
         ..expectData(44)
         ..expectDone()
-        ..expectCancel();
+        ..expectCancel(test.terminate);
     test..listen()
         ..add(42)
         ..add(43)
         ..add(44)
         ..close();
   });
 
   test("pause twice direct-event-unpause", () {
     StreamProtocolTest test = new StreamProtocolTest();
     Completer completer = new Completer();
     test..expectListen()
         ..expectData(42, () {
             test.pause();
             test.pause(completer.future);
             test.add(43);
             test.add(44);
             test.close();
           })
         ..expectPause(() {
             completer.future.then((v) => test.resume());
             completer.complete(null);
           })
         ..expectData(43)
         ..expectData(44)
         ..expectDone()
-        ..expectCancel();
+        ..expectCancel(test.terminate);
     test..listen()
         ..add(42);
   });
 }
 
 class TestError { const TestError(); }
 
 testRethrow() {
   TestError error = const TestError();
 
-
   testStream(name, streamValueTransform) {
     test("rethrow-$name-value", () {
       StreamController c = new StreamController();
       Stream s = streamValueTransform(c.stream, (v) { throw error; });
       s.listen((_) { Expect.fail("unexpected value"); }, onError: expectAsync1(
           (e) { Expect.identical(error, e); }));
       c.add(null);
       c.close();
     });
   }
@@ skipping 117 matching lines @@
       test.add(87);
       sub.cancel();
     });
     test.expectCancel(() {
       test.add(37);
       test.terminate();
     });
   });
 }
 
+void testAsBroadcast() {
+  test("asBroadcast-not-canceled", () {
+    StreamProtocolTest test = new StreamProtocolTest.asBroadcast();
+    var sub;
+    test..expectListen()
+        ..expectBroadcastListen((_) {
+            test.add(42);
+          })
+        ..expectData(42, () {
+            sub.cancel();
+          })
+        ..expectBroadcastCancel((_) {
+            sub = test.listen();
+          })
+        ..expectBroadcastListen((_) {
+            test.terminate();
+          });
+    sub = test.listen();
+  });
+
+  test("asBroadcast-canceled", () {
+    StreamProtocolTest test = new StreamProtocolTest.asBroadcast();
+    var sub;
+    test..expectListen()
+        ..expectBroadcastListen((_) {
+            test.add(42);
+          })
+        ..expectData(42, () {
+            sub.cancel();
+          })
+        ..expectBroadcastCancel((originalSub) {
+            originalSub.cancel();
+          })
+        ..expectCancel(test.terminate);
+    sub = test.listen();
+  });
+
+  test("asBroadcast-pause-original", () {
+    StreamProtocolTest test = new StreamProtocolTest.asBroadcast();
+    var sub;
+    test..expectListen()
+        ..expectBroadcastListen((_) {
+            test.add(42);
+            test.add(43);
+          })
+        ..expectData(42, () {
+            sub.cancel();
+          })
+        ..expectBroadcastCancel((originalSub) {
+            originalSub.pause();  // Pause before sending 43 from original sub.
+          })
+        ..expectPause(() {
+            sub = test.listen();
+          })
+        ..expectBroadcastListen((originalSub) {
+            originalSub.resume();
+          })
+        ..expectData(43)
+        ..expectResume(() {
+            test.close();
+          })
+        ..expectDone()
+        ..expectBroadcastCancel()
+        ..expectCancel(test.terminate);
+    sub = test.listen();
+  });
+}
+
 main() {
   testController();
   testSingleController();
   testExtraMethods();
   testPause();
   testRethrow();
   testBroadcastController();
+  testAsBroadcast();
 }