Make StreamSubscription be the active part of a stream.

pkg/scheduled_test/lib/src/mock_clock.dart

Index: pkg/scheduled_test/lib/src/mock_clock.dart
diff --git a/pkg/scheduled_test/lib/src/mock_clock.dart b/pkg/scheduled_test/lib/src/mock_clock.dart
index 6abe81f155031ca75d4c4b68114b4b9eb725c2d1..c80a9552661329cf8e4402afdb4bddf75652d473 100644
--- a/pkg/scheduled_test/lib/src/mock_clock.dart
+++ b/pkg/scheduled_test/lib/src/mock_clock.dart
@@ -44,27 +44,43 @@ class Clock {
   int get time => _time;
   int _time = 0;
 
+  /// Collection of controllers of all subscribed listeners.
+  ///
+  /// [StreamController] is not overriding [Object.operator==], so this is
+  /// effectively an identity map.
+  Set<StreamController> _subscriptions = new Set<StreamController>();

[floitsch, 2013/05/24 15:53:17]

Add TODO that we want to replace this with something more efficient.

+
+  Clock._();
+
   /// The stream of millisecond ticks of the clock.
   Stream<int> get onTick {
-    if (_onTickControllerStream == null) {
-      _onTickControllerStream = _onTickController.stream.asBroadcastStream();
-    }
-    return _onTickControllerStream;
+    StreamController<int> controller;
+    controller = new StreamController<int>(
+        onListen: () {
+          _subscriptions.add(controller);
+        },
+        onCancel: () {
+          _subscriptions.remove(controller);
+        });
+    return controller.stream;

[nweiz, 2013/05/24 20:12:31]

I'd rather see this pattern generalized, so it's clearer what's going on and can
be used elsewhere if necessary.

One possible way to do this would be to have a wrapper class like
"UncancellableBroadcastStream" that internally ensures that there's always a
dummy subscription on the wrapped stream so it doesn't get completely cancelled.

[floitsch, 2013/05/24 20:40:48]

https://codereview.chromium.org/16003002/

[nweiz, 2013/05/24 20:49:19]

That looks good. Whichever of this CL or that CL that gets submitted second
should change this code to use that constructor.

   }
 
-  final _onTickController = new StreamController<int>();
-  Stream<int> _onTickControllerStream;
-
-  Clock._();
 
   /// Advances the clock forward by [milliseconds]. This works like synchronous
   /// code that takes [milliseconds] to execute; any [Timer]s that are scheduled
   /// to fire during the interval will do so asynchronously once control returns
   /// to the event loop.
-  void tick([int milliseconds=1]) {
+  void tick([int milliseconds = 1]) {

[nweiz, 2013/05/24 20:12:31]

Our style is to omit spaces around "=" in default value assignment.

Alternately, omit the default value entirely and do "if (milliseconds == null)
milliseconds = 1" in the body.

     for (var i = 0; i < milliseconds; i++) {
       var tickTime = ++_time;
-      new Future.value().then((_) => _onTickController.add(tickTime));
+      runAsync(() {
+        List<StreamController> controllers = _subscriptions.toList();
+        for (StreamController controller in controllers) {
+          if (_subscriptions.contains(controller)) {
+            controller.add(tickTime);
+          }
+        }
+      });
     }
   }
 
@@ -74,7 +90,7 @@ class Clock {
   /// code runs before the next tick.
   void run() {
     pumpEventQueue().then((_) {
-      if (!_onTickController.hasListener) return;
+      if (_subscriptions.isEmpty) return;
       tick();
       return run();
     });