Reapply "Active stream subscriptions".

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..46a37afcf2e246453071150d5b44fac314843f67 100644
--- a/pkg/scheduled_test/lib/src/mock_clock.dart
+++ b/pkg/scheduled_test/lib/src/mock_clock.dart
@@ -44,27 +44,25 @@ class Clock {
   int get time => _time;
   int _time = 0;
 
-  /// The stream of millisecond ticks of the clock.
-  Stream<int> get onTick {
-    if (_onTickControllerStream == null) {
-      _onTickControllerStream = _onTickController.stream.asBroadcastStream();
-    }
-    return _onTickControllerStream;
-  }
-
-  final _onTickController = new StreamController<int>();
-  Stream<int> _onTickControllerStream;
+  /// Controller providing streams for listening.
+  StreamController<int> _multiplexController =
+      new StreamController<int>.multiplex();
 
   Clock._();
 
+  /// The stream of millisecond ticks of the clock.
+  Stream<int> get onTick => _multiplexController.stream;
+
   /// 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]) {
     for (var i = 0; i < milliseconds; i++) {
       var tickTime = ++_time;
-      new Future.value().then((_) => _onTickController.add(tickTime));
+      runAsync(() {
+        _multiplexController.add(tickTime);
+      });
     }
   }
 
@@ -74,7 +72,7 @@ class Clock {
   /// code runs before the next tick.
   void run() {
     pumpEventQueue().then((_) {
-      if (!_onTickController.hasListener) return;
+      if (!_multiplexController.hasListener) return;
       tick();
       return run();
     });