add Stream.toAsyncIterableRuntime api (#4680)

Author: tim-smart

.changeset/funny-islands-relate.md

@@ -0,0 +1,18 @@
+---
+"effect": minor
+---
+
+add Stream.toAsyncIterable\* apis
+
+```ts
+import { Stream } from "effect"
+
+// Will print:
+// 1
+// 2
+// 3
+const stream = Stream.make(1, 2, 3)
+for await (const result of Stream.toAsyncIterable(stream)) {
+  console.log(result)
+}
+```

packages/effect/src/Stream.ts

@@ -5332,6 +5332,34 @@ export const toReadableStreamRuntime: {
   ): ReadableStream<A>
 } = internal.toReadableStreamRuntime
 
+/**
+ * Converts the stream to a `AsyncIterable` using the provided runtime.
+ *
+ * @since 3.15.0
+ * @category destructors
+ */
+export const toAsyncIterableRuntime: {
+  <A, XR>(runtime: Runtime<XR>): <E, R extends XR>(self: Stream<A, E, R>) => AsyncIterable<A>
+  <A, E, XR, R extends XR>(self: Stream<A, E, R>, runtime: Runtime<XR>): AsyncIterable<A>
+} = internal.toAsyncIterableRuntime
+
+/**
+ * Converts the stream to a `AsyncIterable` capturing the required dependencies.
+ *
+ * @since 3.15.0
+ * @category destructors
+ */
+export const toAsyncIterableEffect: <A, E, R>(self: Stream<A, E, R>) => Effect.Effect<AsyncIterable<A>, never, R> =
+  internal.toAsyncIterableEffect
+
+/**
+ * Converts the stream to a `AsyncIterable`.
+ *
+ * @since 3.15.0
+ * @category destructors
+ */
+export const toAsyncIterable: <A, E>(self: Stream<A, E>) => AsyncIterable<A> = internal.toAsyncIterable
+
 /**
  * Applies the transducer to the stream and emits its outputs.
  *

packages/effect/src/internal/stream.ts

@@ -7200,6 +7200,74 @@ export const transduce = dual<
   }
 )
 
+/** @internal */
+export const toAsyncIterableRuntime = dual<
+  <A, XR>(
+    runtime: Runtime.Runtime<XR>
+  ) => <E, R extends XR>(self: Stream.Stream<A, E, R>) => AsyncIterable<A>,
+  <A, E, XR, R extends XR>(
+    self: Stream.Stream<A, E, R>,
+    runtime: Runtime.Runtime<XR>
+  ) => AsyncIterable<A>
+>(
+  (args) => isStream(args[0]),
+  <A, E, XR, R extends XR>(
+    self: Stream.Stream<A, E, R>,
+    runtime: Runtime.Runtime<XR>
+  ): AsyncIterable<A> => {
+    const runFork = Runtime.runFork(runtime)
+    return {
+      [Symbol.asyncIterator]() {
+        let currentResolve: ((value: IteratorResult<A>) => void) | undefined = undefined
+        let currentReject: ((reason: any) => void) | undefined = undefined
+        let fiber: Fiber.RuntimeFiber<void, E> | undefined = undefined
+        const latch = Effect.unsafeMakeLatch(false)
+        return {
+          next() {
+            if (!fiber) {
+              fiber = runFork(runForEach(self, (value) =>
+                latch.whenOpen(Effect.sync(() => {
+                  latch.unsafeClose()
+                  currentResolve!({ done: false, value })
+                  currentResolve = currentReject = undefined
+                }))))
+              fiber.addObserver((exit) => {
+                fiber = Effect.runFork(latch.whenOpen(Effect.sync(() => {
+                  if (exit._tag === "Failure") {
+                    currentReject!(Cause.squash(exit.cause))
+                  } else {
+                    currentResolve!({ done: true, value: void 0 })
+                  }
+                  currentResolve = currentReject = undefined
+                })))
+              })
+            }
+            return new Promise<IteratorResult<A>>((resolve, reject) => {
+              currentResolve = resolve
+              currentReject = reject
+              latch.unsafeOpen()
+            })
+          },
+          return() {
+            if (!fiber) return Promise.resolve({ done: true, value: void 0 })
+            return Effect.runPromise(Effect.as(Fiber.interrupt(fiber), { done: true, value: void 0 }))
+          }
+        }
+      }
+    }
+  }
+)
+
+/** @internal */
+export const toAsyncIterable = <A, E>(self: Stream.Stream<A, E>): AsyncIterable<A> =>
+  toAsyncIterableRuntime(self, Runtime.defaultRuntime)
+
+/** @internal */
+export const toAsyncIterableEffect = <A, E, R>(
+  self: Stream.Stream<A, E, R>
+): Effect.Effect<AsyncIterable<A>, never, R> =>
+  Effect.map(Effect.runtime<R>(), (runtime) => toAsyncIterableRuntime(self, runtime))
+
 /** @internal */
 export const unfold = <S, A>(s: S, f: (s: S) => Option.Option<readonly [A, S]>): Stream.Stream<A> =>
   unfoldChunk(s, (s) => pipe(f(s), Option.map(([a, s]) => [Chunk.of(a), s])))

packages/effect/test/Stream/conversions.test.ts

@@ -73,4 +73,13 @@ describe("Stream", () => {
       )
       deepStrictEqual(queue, Exit.die(new Cause.RuntimeException("die")))
     }))
+
+  it("toAsyncIterable", async () => {
+    const stream = Stream.make(1, 2, 3)
+    const results: Array<number> = []
+    for await (const result of Stream.toAsyncIterable(stream)) {
+      results.push(result)
+    }
+    deepStrictEqual(results, [1, 2, 3])
+  })
 })