Merge pull request #1016 from thefrontside/interval-stream-v4

✨ add interval() helper to consume a stream of intervals, port from v3

Author: cowboyd

lib/interval.ts

@@ -0,0 +1,31 @@
+import { resource } from "./resource.ts";
+import { createSignal } from "./signal.ts";
+import type { Stream } from "./types.ts";
+
+/**
+ * Consume an interval as an infinite stream.
+ *
+ * ```ts
+ * let startTime = Date.now();
+ *
+ * for (let _ of yield* each(interval(10))) {
+ *   let elapsed = Date.now() - startTime;
+ *   console.log(`elapsed time: ${elapsed} ms`);
+ *   yield* each.next();
+ * }
+ * ```
+ * @param milliseconds - how long to delay between each item in the stream
+ */
+export function interval(milliseconds: number): Stream<void, never> {
+  return resource(function* (provide) {
+    let signal = createSignal<void, never>();
+
+    let id = setInterval(signal.send, milliseconds);
+
+    try {
+      yield* provide(yield* signal);
+    } finally {
+      clearInterval(id);
+    }
+  });
+}

lib/mod.ts

@@ -5,6 +5,7 @@ export * from "./context.ts";
 export * from "./scope.ts";
 export * from "./suspend.ts";
 export * from "./sleep.ts";
+export * from "./interval.ts";
 export * from "./run.ts";
 export * from "./spawn.ts";
 export * from "./resource.ts";

test/interval.test.ts

@@ -0,0 +1,27 @@
+import { call, each, interval, race, run, sleep, spawn } from "../mod.ts";
+import { describe, expect, it } from "./suite.ts";
+
+describe("interval", () => {
+  it("can be iterated", async () => {
+    await run(function* () {
+      let task = yield* spawn(function* () {
+        let total = 0;
+        for (let _ of yield* each(interval(1))) {
+          total++;
+          if (total == 10) {
+            return total;
+          }
+          yield* each.next();
+        }
+      });
+      let result = yield* race([
+        task,
+        call(function* () {
+          yield* sleep(50);
+          return "interval not producing!";
+        }),
+      ]);
+      expect(result).toEqual(10);
+    });
+  });
+});