Add Stream.periodic.

sdk/lib/async/stream.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.
 
 part of dart.async;
 
 // -------------------------------------------------------------------
 // Core Stream types
 // -------------------------------------------------------------------
 
@@ skipping 65 matching lines @@
 
   /**
    * Creates a single-subscription stream that gets its data from [data].
    */
   factory Stream.fromIterable(Iterable<T> data) {
     _PendingEvents iterableEvents = new _IterablePendingEvents<T>(data);
     return new _GeneratedSingleStreamImpl<T>(iterableEvents);
   }
 
   /**
+   * Creates a stream that repeatedly emits events at [period] intervals.
+   *
+   * The event values are computed by invoking [computation]. The argument to
+   * this callback is an integer that starts with 0 and is incremented for
+   * every event.
+   *
+   * If [computation] is omitted the event values will all be `null`.
+   */
+  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();
+            timer = null;
+          }
+        });
+    return controller.stream;
+  }
+
+  /**
    * Reports whether this stream is a broadcast stream.
    */
   bool get isBroadcast => false;
 
   /**
    * Returns a multi-subscription stream that produces the same events as this.
    *
    * If this stream is single-subscription, return a new stream that allows
    * multiple subscribers. It will subscribe to this stream when its first
    * subscriber is added, and unsubscribe again when the last subscription is
@@ skipping 1031 matching lines @@
 
 /* TODO(8997): Implement EventSink instead, */
 class _EventOutputSinkWrapper<T> extends StreamSink<T> {
   _EventOutputSink _sink;
   _EventOutputSinkWrapper(this._sink);
 
   void add(T data) { _sink._sendData(data); }
   void addError(AsyncError error) { _sink._sendError(error); }
   void close() { _sink._sendDone(); }
 }