Add Stream.periodic.

sdk/lib/async/stream.dart

Index: sdk/lib/async/stream.dart
diff --git a/sdk/lib/async/stream.dart b/sdk/lib/async/stream.dart
index 4814427b0690e4656cd345eb7654a24a2f11198a..53840acbb41e6dd219a0009484c822b1a1e11b76 100644
--- a/sdk/lib/async/stream.dart
+++ b/sdk/lib/async/stream.dart
@@ -77,6 +77,68 @@ abstract class Stream<T> {
   }
 
   /**
+   * Creates a stream that repeatedly emits events at [period] intervals.
+   *
+   * The sent event is a [Duration] object that represents the time between

[Sean Eagan, 2013/03/04 17:02:14]

Why give the actual Duration for Stream.periodic but not for Future.delayed
which can also have a longer Duration if the system is busy?

Maybe give both an optional "compute()" argument which by default returns the
actual Duration waited.

[floitsch, 2013/03/14 13:27:34]

Changed to be consistent. Both emit "null" now unless there is a compute
function.

+   * this and the last event. The duration can be both longer than

[Lasse Reichstein Nielsen, 2013/03/04 08:43:04]

last -> previous.
Add "and shorter" to match the "both" (and it is true).

[floitsch, 2013/03/13 19:26:55]

I don't think it can be shorter anymore. Removed the "both".

+   * the initial [period] if the machine was not able to
+   * execute the event early enough (for example, due to load), or if the
+   * Stream had been paused.

[Lasse Reichstein Nielsen, 2013/03/04 08:43:04]

had been -> was

[floitsch, 2013/03/13 19:26:55]

Done.

+   *
+   * The first event is emitted after waiting for the initial period.

[Lasse Reichstein Nielsen, 2013/03/04 08:43:04]

After what? The stream being created or the first subscriber?

[floitsch, 2013/03/13 19:26:55]

reworded.

+   */
+  factory Stream.periodic(Duration period) {
+    Timer timer;
+    StreamController controller;

[Lasse Reichstein Nielsen, 2013/03/04 08:43:04]

Consider making a specialized subclass of _SingleStreamImpl for this instead.

[floitsch, 2013/03/13 19:26:55]

As discussed: keeping it as is. Do you want me to add a todo?

[Lasse Reichstein Nielsen, 2013/03/14 06:54:46]

No, it's fine. We would only be saving some extra object creations, and it would
probably be premature optimization.

+    // userWatch counts the time that the Stream was running (and not paused).
+    Stopwatch userWatch = new Stopwatch();
+    // realWatch counts the actual time it took for the event.
+    Stopwatch realWatch = new Stopwatch();
+
+    void sendEvent() {
+      Duration elapsed = realWatch.elapsed;
+      realWatch.reset();
+      userWatch.reset();
+      controller.add(elapsed);
+    }
+
+    void startRepeatingTimer() {
+      assert(timer == null);
+      timer = new Timer.repeating(period, (Timer timer) {
+        sendEvent();
+      });
+    }
+
+    controller = new StreamController(
+        onPauseStateChange: () {
+          if (controller.isPaused) {
+            timer.cancel();
+            timer = null;
+            userWatch.stop();
+          } else {
+            assert(timer == null);

[Lasse Reichstein Nielsen, 2013/03/04 08:43:04]

Why not fire immediately if realWatch.elapsedMilliseconds >=
period.milliseconds, or set remaining timer based on that instead of userWatch?
You can drop userWatch then.

That is, if the receiver pauses, and then resumes because he's ready again,
before the period has elapsed, why should he not get the event on time? Imagine
something that asks for a period of 5 seconds, then pauses for one second while
executing the event (because it *might* take longer)?

Basically, I want a stream that gives me an event every period, or as soon as
possible after that (in case it's paused or busy at that time.If you pause for
more than period time, you can skip an event, but don't delay unless necessary.

[floitsch, 2013/03/13 19:26:55]

As discussed: keeping the current semantics.

+            Duration elapsed = userWatch.elapsed;
+            userWatch.start();
+            timer = new Timer(period - elapsed, () {
+              timer = null;
+              startRepeatingTimer();
+              sendEvent();
+            });
+          }
+        },
+        onSubscriptionStateChange: () {
+          if (controller.hasSubscribers) {
+            realWatch.start();
+            userWatch.start();
+            startRepeatingTimer();
+          } else {
+            if (timer != null) timer.cancel();
+          }
+        });
+    return controller.stream;
+  }
+
+  /**
    * Reports whether this stream is a broadcast stream.
    */
   bool get isBroadcast => false;