Make StreamSubscription be the active part of a stream.

pkg/scheduled_test/lib/src/mock_clock.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.
 
 /// A library that wraps [Timer] in a way that can be mocked out in test code.
 /// Application code only needs to use [newTimer] to get an instance of [Timer].
 /// Then test code can call [mock] to mock out all new [Timer] instances so that
 /// they're controllable by a returned [Clock] object.
 library mock_clock;
 
@@ skipping 27 matching lines @@
 
 /// A clock that controls when mocked [Timer]s move forward in time. It starts
 /// at time 0 and advances forward millisecond-by-millisecond, broadcasting each
 /// tick on the [onTick] stream.
 class Clock {
   /// The current time of the clock, in milliseconds. Starts at 0.
   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;
+  Stream<int> get onTick => _onTickControllerStream;
+
+  StreamController<int> _onTickController;
+  Stream<int> _onTickControllerStream;
+
+  Clock._() {
+    _initStream();
   }
 
-  final _onTickController = new StreamController<int>();
-  Stream<int> _onTickControllerStream;
-
-  Clock._();
+  /// 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
   /// to fire during the interval will do so asynchronously once control returns
   /// to the event loop.
   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));
     }
   }
 
   /// Automatically progresses forward in time as long as there are still
   /// subscribers to [onTick] (that is, [Timer]s waiting to fire). After each
   /// tick, this pumps the event loop repeatedly so that all non-clock-dependent
   /// 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();
     });
   }
 }
 
 /// A mock implementation of [Timer] that uses [Clock] to keep time, rather than
 /// the system clock.
 class _MockTimer implements Timer {
   /// The time at which the timer should fire.
   final int _time;
 
   /// The callback to run when the timer fires.
   final TimerCallback _callback;
 
   /// The subscription to the [Clock.onTick] stream.
   StreamSubscription _subscription;
 
   _MockTimer(Duration duration, this._callback)
       : _time = _clock.time + duration.inMilliseconds {
     _subscription = _clock.onTick.listen((time) {
       if (time < _time) return;
       _subscription.cancel();
       _callback();
     });
   }
 
   void cancel() => _subscription.cancel();
 }