✨ Iterate Subscriptions with each()

A pesistent gotcha that folks have with iterating streams is that
there can arise a race condition where you want to start consuming a
stream, but items have already been sent to it by the time your
consumer fires up. The fix is to capture a live subscription _before_
your consumer loop fires up. But this leaves you in a pickle because
now you can't use the convenient `each()` helper to do your iteration.

This allows the argument to `each()` to either be a stream or a
subscription. That way, you can have the best of both worlds.

Author: cowboyd

lib/each.ts

@@ -1,6 +1,7 @@
+import { spawn } from "./spawn.ts";
+import { constant } from "./constant.ts";
 import { createContext } from "./context.ts";
 import { useScope } from "./scope.ts";
-import { spawn } from "./spawn.ts";
 import type { Operation, Stream, Subscription } from "./types.ts";
 import { withResolvers } from "./with-resolvers.ts";
 
@@ -28,9 +29,16 @@ import { withResolvers } from "./with-resolvers.ts";
  * @param stream - the stream to iterate
  * @returns an operation to iterate `stream`
  */
-export function each<T>(stream: Stream<T, unknown>): Operation<Iterable<T>> {
+export function each<T>(
+  enumerable: Stream<T, unknown> | Subscription<T, unknown>,
+): Operation<Iterable<T>> {
   return {
     *[Symbol.iterator]() {
+      let stream = typeof (enumerable as Subscription<T, unknown>).next ===
+          "function"
+        ? constant(enumerable as Subscription<T, unknown>)
+        : enumerable as Stream<T, unknown>;
+
       let scope = yield* useScope();
       if (!scope.hasOwn(EachStack)) {
         scope.set(EachStack, []);

test/each.test.ts

@@ -1,5 +1,6 @@
 import { describe, expect, it } from "./suite.ts";
 import {
+  createQueue,
   each,
   type Operation,
   resource,
@@ -54,6 +55,26 @@ describe("each", () => {
     });
   });
 
+  it("can iterate subscriptions", async () => {
+    await run(function* () {
+      let queue = createQueue<number, void>();
+
+      queue.add(1);
+      queue.add(2);
+      queue.add(3);
+      queue.close();
+
+      let items = [];
+
+      for (let num of yield* each(queue)) {
+        items.push(num);
+        yield* each.next();
+      }
+
+      expect(items).toEqual([1, 2, 3]);
+    });
+  });
+
   it("handles context correctly if you break out of a loop", async () => {
     await expect(run(function* () {
       let seq = sequence("hello world");