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 59 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 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;
 
   /**
    * 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 1016 matching lines @@
 }
 
 class _StreamOutputSinkWrapper<T> implements StreamSink<T> {
   _StreamOutputSink _sink;
   _StreamOutputSinkWrapper(this._sink);
 
   void add(T data) => _sink._sendData(data);
   void signalError(AsyncError error) => _sink._sendError(error);
   void close() => _sink._sendDone();
 }