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..db7c2de9f376a8e83b690ee64070d2c5e559e30f 100644
--- a/pkg/scheduled_test/lib/src/mock_clock.dart
+++ b/pkg/scheduled_test/lib/src/mock_clock.dart
@@ -45,17 +45,23 @@ class Clock {
   int _time = 0;
 
   /// The stream of millisecond ticks of the clock.
-  Stream<int> get onTick {
-    if (_onTickControllerStream == null) {
-      _onTickControllerStream = _onTickController.stream.asBroadcastStream();
-    }
-    return _onTickControllerStream;
-  }
+  Stream<int> get onTick => _onTickControllerStream;
 
-  final _onTickController = new StreamController<int>();
+  StreamController<int> _onTickController;
   Stream<int> _onTickControllerStream;
 
-  Clock._();
+  Clock._() {
+    _initStream();
+  }
+
+  /// Creates a new stream controller and its stream as broadcast stream.

[floitsch, 2013/05/22 16:26:29]

The behavior is not the same anymore.
When there are pending events the new version discards pending events.
Make sure the new behavior is acceptable.

+  /// When a stream controller is cancelled, a new controller is created
+  /// to give the perception of a stream that can be listened to more than
+  /// once.
+  void _initStream() {
+    _onTickController = new StreamController<int>(onCancel: _initStream);
+    _onTickControllerStream = _onTickController.stream.asBroadcastStream();
+  }

[nweiz, 2013/05/22 18:42:41]

I don't understand these changes. Why are we creating a new stream when a user
cancels their subscription? What happens to other users with subscriptions to
the old stream?

[Lasse Reichstein Nielsen, 2013/05/24 06:02:49]

The stream-controller only received the cancel when all subscribers of the
broadcast stream are gone.
At that point, the stream is dead, and any new subscribers will have to listen
on a new stream.

I think this should instead be done with one controller per listener. You don't
get events that fires before you listen, you get all events that fire while you
listen, and then your stream is dead when you stop listening.
I've implemented this. It is similar (at least in spirit) to how DOM streams are
implemented - each time you read the .onTick property, you get a new stream.

 
   /// Advances the clock forward by [milliseconds]. This works like synchronous
   /// code that takes [milliseconds] to execute; any [Timer]s that are scheduled
@@ -64,7 +70,7 @@ class Clock {
   void tick([int milliseconds=1]) {
     for (var i = 0; i < milliseconds; i++) {
       var tickTime = ++_time;
-      new Future.value().then((_) => _onTickController.add(tickTime));
+      runAsync(() => _onTickController.add(tickTime));
     }
   }
 
@@ -74,7 +80,7 @@ class Clock {
   /// code runs before the next tick.
   void run() {
     pumpEventQueue().then((_) {
-      if (!_onTickController.hasListener) return;
+      if (_onTickController == null) return;

[floitsch, 2013/05/22 16:26:29]

the _onTickController can never be null. I think the old code should be fine.

[Lasse Reichstein Nielsen, 2013/05/24 06:02:49]

Good point.

       tick();
       return run();
     });