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 8ac0541908681a94f19156baf1051b542fafbdaa..b3f0b36d2081f1e5772d59c7cda721d68b3a21c8 100644
--- a/sdk/lib/async/stream.dart
+++ b/sdk/lib/async/stream.dart
@@ -83,6 +83,69 @@ abstract class Stream<T> {
   }
 
   /**
+   * Creates a stream that repeatedly emits events at [period] intervals.
+   *
+   * The events are computed by invoking [computation]. The argument to
+   * this callback is an integer that starts with 0 and is incremented for
+   * every event.
+   *
+   * By default, the computation always returns `null`.

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

Above: The event values are computed ....
and here: If [computation] is omitted, the event values will all be `null`.

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

Done.

+   *
+   * When a listener subscribes the stream starts by waiting for the given
+   * [period]. After this duration has elapsed it emits the first event.

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

When will the next event be emitted? [period] after the first one was emitted,
not counting time where the stream was paused?

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

Removed the lines. Not adding anything.

+   */
+  factory Stream.periodic(Duration period,
+                          [T computation(int computationCount)]) {
+    if (computation == null) computation = ((i) => null);
+
+    Timer timer;
+    int computationCount = 0;
+    StreamController<T> controller;
+    // Counts the time that the Stream was running (and not paused).
+    Stopwatch watch = new Stopwatch();
+
+    void sendEvent() {
+      watch.reset();
+      T data = computation(computationCount++);
+      controller.add(data);
+    }
+
+    void startPeriodicTimer() {
+      assert(timer == null);
+      timer = new Timer.periodic(period, (Timer timer) {
+        sendEvent();
+      });
+    }
+
+    controller = new StreamController<T>(
+        onPauseStateChange: () {
+          if (controller.isPaused) {
+            timer.cancel();
+            timer = null;
+            watch.stop();
+          } else {
+            assert(timer == null);
+            Duration elapsed = watch.elapsed;
+            watch.start();
+            timer = new Timer(period - elapsed, () {
+              timer = null;
+              startPeriodicTimer();
+              sendEvent();
+            });
+          }
+        },
+        onSubscriptionStateChange: () {
+          if (controller.hasSubscribers) {
+            watch.start();
+            startPeriodicTimer();
+          } else {
+            if (timer != null) timer.cancel();

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

set timer to null here?
Otherwise unsubscribing and then resubscribing will hit the assert in
startPeriodicTimer.

Alternatively, call controller.close() which will prevent resubscription.

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

done.

+          }
+        });
+    return controller.stream;
+  }
+
+  /**
    * Reports whether this stream is a broadcast stream.
    */
   bool get isBroadcast => false;