Make asBroadcastStream take two callbacks.

tests/lib/async/stream_state_helper.dart

 // Copyright (c) 2013, 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.
 
 library stream_state_helper;
 
 import "../../../pkg/unittest/lib/unittest.dart";
 import "dart:async";
 import "dart:collection";
 
@@ skipping 28 matching lines @@
     _streamTest._expectError(this, error, action);
   }
 
   void expectDone([void action()]) {
     _streamTest._expectDone(this, action);
   }
 }
 
 class StreamProtocolTest {
   bool trace = false;
-  // If not a broadcast stream, the onComplete is called automatically by
-  // the first onCancel.
-  bool isBroadcast;
+  final bool isBroadcast;
+  final bool isAsBroadcast;
   StreamController _controller;
   Stream _controllerStream;
   // Most recent subscription created. Used as default for pause/resume.
   SubscriptionProtocolTest _latestSubscription;
   List<Event> _expectations = new List<Event>();
   int _nextExpectationIndex = 0;
   int _subscriptionIdCounter = 0;
   Function _onComplete;
 
   StreamProtocolTest.broadcast({ bool sync: false })
-      : isBroadcast = true {
+      : isBroadcast = true, isAsBroadcast = false {
     _controller = new StreamController.broadcast(
         sync: sync,
         onListen: _onListen,
         onCancel: _onCancel);
     _controllerStream = _controller.stream;
     _onComplete = expectAsync0((){
       _onComplete = null;  // Being null marks the test as being complete.
     });
   }
 
-  StreamProtocolTest({ bool broadcast: false, bool sync: false })
-      : isBroadcast = false {
+  StreamProtocolTest({ bool sync: false })
+    : isBroadcast = false, isAsBroadcast = false {
     _controller = new StreamController(
         sync: sync,
         onListen: _onListen,
         onPause: _onPause,
         onResume: _onResume,
         onCancel: _onCancel);
-    if (broadcast) {
-      _controllerStream = _controller.stream.asBroadcastStream();
-    } else {
-      _controllerStream = _controller.stream;
-    }
+    _controllerStream = _controller.stream;
     _onComplete = expectAsync0((){
       _onComplete = null;  // Being null marks the test as being complete.
     });
+  }
+
+  StreamProtocolTest.asBroadcast({ bool sync: false })
+      : isBroadcast = false, isAsBroadcast = true {
+    _controller = new StreamController(
+        sync: sync,
+        onListen: _onListen,
+        onPause: _onPause,
+        onResume: _onResume,
+        onCancel: _onCancel);
+    _controllerStream = _controller.stream.asBroadcastStream(
+        onListen: _onBroadcastListen,
+        onCancel: _onBroadcastCancel);
+    _onComplete = expectAsync0((){
+      _onComplete = null;  // Being null marks the test as being complete.
+    });
   }
 
   // Actions on the stream and controller.
   void add(var data) { _controller.add(data); }
   void error(var error) { _controller.addError(error); }
   void close() { _controller.close(); }
 
   SubscriptionProtocolTest listen({bool cancelOnError : false}) {
     int subscriptionId = _subscriptionIdCounter++;
 
@@ skipping 99 matching lines @@
   }
 
   void _onCancel() {
     if (trace) print("[Cancelled]");
     _withNextExpectation((Event expect) {
       if (!expect.matchCancel()) {
         _fail("Expected: $expect\n"
               "Found: [Cancelled]\n${expect._stackTrace}");
       }
     });
-    if (!isBroadcast) terminate();
+  }
+
+  void _onBroadcastListen(StreamSubscription sub) {
+    if (trace) print("[BroadcastListen]");
+    _withNextExpectation((Event expect) {
+      if (!expect.matchBroadcastListen(sub)) {
+        _fail("Expected: $expect\n"
+              "Found: [BroadcastListen]\n${expect._stackTrace}");
+      }
+    });
+  }
+
+  void _onBroadcastCancel(StreamSubscription sub) {
+    if (trace) print("[BroadcastCancel]");
+    _withNextExpectation((Event expect) {
+      if (!expect.matchBroadcastCancel(sub)) {
+        _fail("Expected: $expect\n"
+              "Found: [BroadcastCancel]\n${expect._stackTrace}");
+      }
+    });
   }
 
   void _withNextExpectation(void action(Event expect)) {
     if (_nextExpectationIndex == _expectations.length) {
       _nextExpectationIndex++;
       action(new MismatchEvent());
     } else {
       Event next = _expectations[_nextExpectationIndex++];
       action(next);
     }
@@ skipping 63 matching lines @@
   }
 
   void expectCancel([void action()]) {
     if (_onComplete == null) {
       _fail("Adding expectation after completing");
     }
     _expectations.add(
           new CancelCallbackEvent(action));
   }
 
+  void expectBroadcastListen([void action(StreamSubscription sub)]) {
+    if (_onComplete == null) {
+      _fail("Adding expectation after completing");
+    }
+    if (!isAsBroadcast) throw new StateError("Not an asBroadcast stream");
+    _expectations.add(new BroadcastListenCallbackEvent(action));
+  }
+
+  void expectBroadcastCancel([void action(StreamSubscription sub)]) {
+    if (_onComplete == null) {
+      _fail("Adding expectation after completing");
+    }
+    if (!isAsBroadcast) throw new StateError("Not an asBroadcast stream");
+    _expectations.add(new BroadcastCancelCallbackEvent(action));
+  }
+
+  void expectBroadcastListenOpt([void action(StreamSubscription sub)]) {
+    if (_onComplete == null) {
+      _fail("Adding expectation after completing");
+    }
+    if (!isAsBroadcast) return;
+    _expectations.add(new BroadcastListenCallbackEvent(action));
+  }
+
+  void expectBroadcastCancelOpt([void action(StreamSubscription sub)]) {
+    if (_onComplete == null) {
+      _fail("Adding expectation after completing");
+    }
+    if (!isAsBroadcast) return;
+    _expectations.add(new BroadcastCancelCallbackEvent(action));
+  }
+
   void _fail(String message) {
     if (_nextExpectationIndex == 0) {
       throw "Unexpected event:\n$message\nNo earlier events matched.";
     }
     StringBuffer buf = new StringBuffer();
     for (int i = 0; i < _expectations.length; i++) {
       if (i == _nextExpectationIndex - 1) {
         buf.write("->");
       } else {
         buf.write("  ");
       }
       buf.write(_expectations[i]);
       buf.write("\n");
     }
     throw "Unexpected event:\n$message\nAll expectations:\n$buf";
   }
 }
 
 class Event {
   Function _action;
   StackTrace _stackTrace;
   Event(void action())
       : _action = (action == null) ? null : expectAsync0(action) {
     try { throw 0; } catch (_, s) { _stackTrace = s; }
   }
+  Event.broadcast(void action(StreamSubscription sub))
+      : _action = (action == null) ? null : expectAsync1(action) {
+    try { throw 0; } catch (_, s) { _stackTrace = s; }
+  }
 
   bool matchData(int id, var data) {
     return false;
   }
 
   bool matchError(int id, e) {
     return false;
   }
 
   bool matchDone(int id) {
@@ skipping 17 matching lines @@
     if (_action != null) _action();
     return true;
   }
 
   bool matchCancel() {
     if (!_testCancel()) return false;
     if (_action != null) _action();
     return true;
   }
 
+  bool matchBroadcastListen(StreamSubscription sub) {
+    if (!_testBroadcastListen()) return false;
+    if (_action != null) _action(sub);
+    return true;
+  }
+
+  bool matchBroadcastCancel(StreamSubscription sub) {
+    if (!_testBroadcastCancel()) return false;
+    if (_action != null) _action(sub);
+    return true;
+  }
+
   bool _testData(_) => false;
   bool _testError(_) => false;
   bool _testDone() => false;
   bool _testPause() => false;
   bool _testResume() => false;
   bool _testSubscribe() => false;
   bool _testCancel() => false;
+  bool _testBroadcastListen() => false;
+  bool _testBroadcastCancel() => false;
 }
 
 class SubscriptionEvent extends Event {
   SubscriptionProtocolTest subscription;
   SubscriptionEvent(this.subscription, void action()) : super(action);
 
   bool matchData(int id, var data) {
     if (subscription != null && subscription.id != id) return false;
     if (!_testData(data)) return false;
     if (_action != null) _action();
@@ skipping 61 matching lines @@
   bool _testSubscribe() => true;
   String toString() => "[Subscribed]";
 }
 
 class CancelCallbackEvent extends Event {
   CancelCallbackEvent(void action()) : super(action);
   bool _testCancel() => true;
   String toString() => "[Cancelled]";
 }
 
+class _BroadcastEventCallback extends Event {
+  Function _action;
+  _BroadcastEventCallback();
+}
+
+class BroadcastCancelCallbackEvent extends Event {
+  BroadcastCancelCallbackEvent(void action(StreamSubscription sub))
+    : super.broadcast(action);
+  bool _testBroadcastCancel() => true;
+  String toString() => "[BroadcastCancel]";
+}
+
+class BroadcastListenCallbackEvent extends Event {
+  BroadcastListenCallbackEvent(void action(StreamSubscription sub))
+    : super.broadcast(action);
+  bool _testBroadcastListen() => true;
+  String toString() => "[BroadcastListen]";
+}
+
 /** Event matcher that matches any other event. */
 class LogAnyEvent extends Event {
   String _actual = "*Not matched yet*";
 
   LogAnyEvent(void action()) : super(action);
 
   bool _testData(var data) {
     _actual = "*[Data $data]";
     return true;
   }
@@ skipping 21 matching lines @@
   bool _testSubcribe() {
     _actual = "*[Subscribed]";
     return true;
   }
 
   bool _testCancel() {
     _actual = "*[Cancelled]";
     return true;
   }
 
+  bool _testBroadcastListen() {
+    _actual = "*[BroadcastListen]";
+    return true;
+  }
+
+  bool _testBroadcastCancel() {
+    _actual = "*[BroadcastCancel]";
+    return true;
+  }
+
   /** Returns a representation of the event it was tested against. */
   String toString() => _actual;
 }