add Promise based apis to Fiber{Handle,Set,Map} modules (#4401)

Effect-TS · Feb 5, 2025 · 2940151 · 2940151
1 parent c0e5bef
commit 2940151
Show file tree

Hide file tree

Showing 7 changed files with 209 additions and 0 deletions.
diff --git a/.changeset/chatty-terms-occur.md b/.changeset/chatty-terms-occur.md
@@ -0,0 +1,5 @@
+---
+"effect": minor
+---
+
+add Promise based apis to Fiber{Handle,Set,Map} modules
diff --git a/packages/effect/src/FiberHandle.ts b/packages/effect/src/FiberHandle.ts
@@ -143,6 +143,25 @@ export const makeRuntime = <R, E = unknown, A = unknown>(): Effect.Effect<
     (self) => runtime(self)<R>()
   )
 
+/**
+ * Create an Effect run function that is backed by a FiberHandle.
+ *
+ * @since 3.13.0
+ * @categories constructors
+ */
+export const makeRuntimePromise = <R = never, A = unknown, E = unknown>(): Effect.Effect<
+  <XE extends E, XA extends A>(
+    effect: Effect.Effect<XA, XE, R>,
+    options?: Runtime.RunForkOptions | undefined
+  ) => Promise<XA>,
+  never,
+  Scope.Scope | R
+> =>
+  Effect.flatMap(
+    make<A, E>(),
+    (self) => runtimePromise(self)<R>()
+  )
+
 const internalFiberIdId = -1
 const internalFiberId = FiberId.make(internalFiberIdId, 0)
 const isInternalInterruption = Cause.reduceWithContext(undefined, {
@@ -436,6 +455,46 @@ export const runtime: <A, E>(
     }
   )
 
+/**
+ * Capture a Runtime and use it to fork Effect's, adding the forked fibers to the FiberHandle.
+ *
+ * The returned run function will return Promise's that will resolve when the
+ * fiber completes.
+ *
+ * @since 3.13.0
+ * @categories combinators
+ */
+export const runtimePromise = <A, E>(self: FiberHandle<A, E>): <R = never>() => Effect.Effect<
+  <XE extends E, XA extends A>(
+    effect: Effect.Effect<XA, XE, R>,
+    options?:
+      | Runtime.RunForkOptions & { readonly propagateInterruption?: boolean | undefined }
+      | undefined
+  ) => Promise<XA>,
+  never,
+  R
+> =>
+<R>() =>
+  Effect.map(
+    runtime(self)<R>(),
+    (runFork) =>
+    <XE extends E, XA extends A>(
+      effect: Effect.Effect<XA, XE, R>,
+      options?:
+        | Runtime.RunForkOptions & { readonly propagateInterruption?: boolean | undefined }
+        | undefined
+    ): Promise<XA> =>
+      new Promise((resolve, reject) =>
+        runFork(effect, options).addObserver((exit) => {
+          if (Exit.isSuccess(exit)) {
+            resolve(exit.value)
+          } else {
+            reject(Cause.squash(exit.cause))
+          }
+        })
+      )
+  )
+
 /**
  * If any of the Fiber's in the handle terminate with a failure,
  * the returned Effect will terminate with the first failure that occurred.

diff --git a/packages/effect/src/FiberMap.ts b/packages/effect/src/FiberMap.ts
@@ -158,6 +158,30 @@ export const makeRuntime = <R, K, E = unknown, A = unknown>(): Effect.Effect<
     (self) => runtime(self)<R>()
   )
 
+/**
+ * Create an Effect run function that is backed by a FiberMap.
+ *
+ * @since 3.13.0
+ * @categories constructors
+ */
+export const makeRuntimePromise = <R, K, A = unknown, E = unknown>(): Effect.Effect<
+  <XE extends E, XA extends A>(
+    key: K,
+    effect: Effect.Effect<XA, XE, R>,
+    options?:
+      | Runtime.RunForkOptions & {
+        readonly onlyIfMissing?: boolean | undefined
+      }
+      | undefined
+  ) => Promise<XA>,
+  never,
+  Scope.Scope | R
+> =>
+  Effect.flatMap(
+    make<K, A, E>(),
+    (self) => runtimePromise(self)<R>()
+  )
+
 const internalFiberIdId = -1
 const internalFiberId = FiberId.make(internalFiberIdId, 0)
 const isInternalInterruption = Cause.reduceWithContext(undefined, {
@@ -539,6 +563,48 @@ export const runtime: <K, A, E>(
     }
   )
 
+/**
+ * Capture a Runtime and use it to fork Effect's, adding the forked fibers to the FiberMap.
+ *
+ * @since 3.13.0
+ * @categories combinators
+ */
+export const runtimePromise = <K, A, E>(self: FiberMap<K, A, E>): <R = never>() => Effect.Effect<
+  <XE extends E, XA extends A>(
+    key: K,
+    effect: Effect.Effect<XA, XE, R>,
+    options?:
+      | Runtime.RunForkOptions & {
+        readonly onlyIfMissing?: boolean | undefined
+        readonly propagateInterruption?: boolean | undefined
+      }
+      | undefined
+  ) => Promise<XA>,
+  never,
+  R
+> =>
+<R>() =>
+  Effect.map(
+    runtime(self)<R>(),
+    (runFork) =>
+    <XE extends E, XA extends A>(
+      key: K,
+      effect: Effect.Effect<XA, XE, R>,
+      options?:
+        | Runtime.RunForkOptions & { readonly propagateInterruption?: boolean | undefined }
+        | undefined
+    ): Promise<XA> =>
+      new Promise((resolve, reject) =>
+        runFork(key, effect, options).addObserver((exit) => {
+          if (Exit.isSuccess(exit)) {
+            resolve(exit.value)
+          } else {
+            reject(Cause.squash(exit.cause))
+          }
+        })
+      )
+  )
+
 /**
  * @since 2.0.0
  * @categories combinators

diff --git a/packages/effect/src/FiberSet.ts b/packages/effect/src/FiberSet.ts
@@ -146,6 +146,25 @@ export const makeRuntime = <R = never, A = unknown, E = unknown>(): Effect.Effec
     (self) => runtime(self)<R>()
   )
 
+/**
+ * Create an Effect run function that is backed by a FiberSet.
+ *
+ * @since 3.13.0
+ * @categories constructors
+ */
+export const makeRuntimePromise = <R = never, A = unknown, E = unknown>(): Effect.Effect<
+  <XE extends E, XA extends A>(
+    effect: Effect.Effect<XA, XE, R>,
+    options?: Runtime.RunForkOptions | undefined
+  ) => Promise<XA>,
+  never,
+  Scope.Scope | R
+> =>
+  Effect.flatMap(
+    make<A, E>(),
+    (self) => runtimePromise(self)<R>()
+  )
+
 const internalFiberIdId = -1
 const internalFiberId = FiberId.make(internalFiberIdId, 0)
 const isInternalInterruption = Cause.reduceWithContext(undefined, {
@@ -375,6 +394,45 @@ export const runtime: <A, E>(
     }
   )
 
+/**
+ * Capture a Runtime and use it to fork Effect's, adding the forked fibers to the FiberSet.
+ *
+ * The returned run function will return Promise's.
+ *
+ * @since 3.13.0
+ * @categories combinators
+ */
+export const runtimePromise = <A, E>(self: FiberSet<A, E>): <R = never>() => Effect.Effect<
+  <XE extends E, XA extends A>(
+    effect: Effect.Effect<XA, XE, R>,
+    options?:
+      | Runtime.RunForkOptions & { readonly propagateInterruption?: boolean | undefined }
+      | undefined
+  ) => Promise<XA>,
+  never,
+  R
+> =>
+<R>() =>
+  Effect.map(
+    runtime(self)<R>(),
+    (runFork) =>
+    <XE extends E, XA extends A>(
+      effect: Effect.Effect<XA, XE, R>,
+      options?:
+        | Runtime.RunForkOptions & { readonly propagateInterruption?: boolean | undefined }
+        | undefined
+    ): Promise<XA> =>
+      new Promise((resolve, reject) =>
+        runFork(effect, options).addObserver((exit) => {
+          if (Exit.isSuccess(exit)) {
+            resolve(exit.value)
+          } else {
+            reject(Cause.squash(exit.cause))
+          }
+        })
+      )
+  )
+
 /**
  * @since 2.0.0
  * @categories combinators

diff --git a/packages/effect/test/FiberHandle.test.ts b/packages/effect/test/FiberHandle.test.ts
@@ -111,4 +111,11 @@ describe("FiberHandle", () => {
       yield* TestClock.adjust(500)
       assert.isDefined(fiber.unsafePoll())
     }))
+
+  it.scoped("makeRuntimePromise", () =>
+    Effect.gen(function*() {
+      const run = yield* FiberHandle.makeRuntimePromise()
+      const result = yield* Effect.promise(() => run(Effect.succeed("done")))
+      strictEqual(result, "done")
+    }))
 })
diff --git a/packages/effect/test/FiberMap.test.ts b/packages/effect/test/FiberMap.test.ts
@@ -136,4 +136,11 @@ describe("FiberMap", () => {
       yield* TestClock.adjust(500)
       assert.isDefined(fiber.unsafePoll())
     }))
+
+  it.scoped("makeRuntimePromise", () =>
+    Effect.gen(function*() {
+      const run = yield* FiberMap.makeRuntimePromise<never, string>()
+      const result = yield* Effect.promise(() => run("a", Effect.succeed("done")))
+      strictEqual(result, "done")
+    }))
 })
diff --git a/packages/effect/test/FiberSet.test.ts b/packages/effect/test/FiberSet.test.ts
@@ -108,4 +108,11 @@ describe("FiberSet", () => {
       yield* TestClock.adjust(500)
       assert.isDefined(fiber.unsafePoll())
     }))
+
+  it.scoped("makeRuntimePromise", () =>
+    Effect.gen(function*() {
+      const run = yield* FiberSet.makeRuntimePromise()
+      const result = yield* Effect.promise(() => run(Effect.succeed("done")))
+      strictEqual(result, "done")
+    }))
 })