Consolidate Effect and Stream async methods (#2087)

Author: IMax153

.changeset/cuddly-parrots-know.md

@@ -0,0 +1,23 @@
+---
+"effect": minor
+---
+
+Consolidate `Effect.asyncOption`, `Effect.asyncEither`, `Stream.asyncOption`, `Stream.asyncEither`, and `Stream.asyncInterrupt`
+
+This PR removes `Effect.asyncOption` and `Effect.asyncEither` as their behavior can be entirely implemented with the new signature of `Effect.async`, which optionally returns a cleanup `Effect` from the registration callback.
+
+```ts
+declare const async: <A, E = never, R = never>(
+  register: (callback: (_: Effect<A, E, R>) => void, signal: AbortSignal) => void | Effect<void, never, R>,
+  blockingOn?: FiberId
+) => Effect<A, E, R>
+```
+
+Additionally, this PR removes `Stream.asyncOption`, `Stream.asyncEither`, and `Stream.asyncInterrupt` as their behavior can be entirely implemented with the new signature of `Stream.async`, which can optionally return a cleanup `Effect` from the registration callback.
+
+```ts
+declare const async: <A, E = never, R = never>(
+  register: (emit: Emit<R, E, A, void>) => Effect<void, never, R> | void,
+  outputBuffer?: number
+) => Stream<A, E, R>
+```

package.json

@@ -1,7 +1,7 @@
 {
   "private": true,
   "type": "module",
-  "packageManager": "pnpm@8.12.1",
+  "packageManager": "pnpm@8.15.1",
   "workspaces": [
     "packages/*"
   ],

packages/effect/src/Effect.ts

@@ -976,17 +976,20 @@ export const validateFirst: {
 // -------------------------------------------------------------------------------------
 
 /**
- * Imports an asynchronous side-effect into a pure `Effect` value.
- * The callback function `Effect<A, E, R> => void` must be called at most once.
+ * Imports an asynchronous side-effect into a pure `Effect` value. The callback
+ * function `Effect<A, E, R> => void` **MUST** be called at most once.
  *
- * If an Effect is returned by the registration function, it will be executed
- * if the fiber executing the effect is interrupted.
+ * The registration function can optionally return an Effect, which will be
+ * executed if the `Fiber` executing this Effect is interrupted.
  *
  * The registration function can also receive an `AbortSignal` if required for
  * interruption.
  *
- * The `FiberId` of the fiber that may complete the async callback may be
- * provided to allow for better diagnostics.
+ * The `FiberId` of the fiber that may complete the async callback may also be
+ * specified. This is called the "blocking fiber" because it suspends the fiber
+ * executing the `async` Effect (i.e. semantically blocks the fiber from making
+ * progress). Specifying this fiber id in cases where it is known will improve
+ * diagnostics, but not affect the behavior of the returned effect.
  *
  * @since 2.0.0
  * @category constructors
@@ -1005,51 +1008,9 @@ export const async: <A, E = never, R = never>(
  * @since 2.0.0
  * @category constructors
  */
-export const asyncEffect: <A, E, R, X, E2, R2>(
-  register: (callback: (_: Effect<A, E, R>) => void) => Effect<X, E2, R2>
-) => Effect<A, E | E2, R | R2> = _runtime.asyncEffect
-
-/**
- * Imports an asynchronous effect into a pure `Effect` value, possibly returning
- * the value synchronously.
- *
- * If the register function returns a value synchronously, then the callback
- * function `Effect<A, E, R> => void` must not be called. Otherwise the callback
- * function must be called at most once.
- *
- * The `FiberId` of the fiber that may complete the async callback may be
- * provided to allow for better diagnostics.
- *
- * @since 2.0.0
- * @category constructors
- */
-export const asyncOption: <A, E = never, R = never>(
-  register: (callback: (_: Effect<A, E, R>) => void) => Option.Option<Effect<A, E, R>>,
-  blockingOn?: FiberId.FiberId
-) => Effect<A, E, R> = effect.asyncOption
-
-/**
- * Imports an asynchronous side-effect into an effect. It has the option of
- * returning the value synchronously, which is useful in cases where it cannot
- * be determined if the effect is synchronous or asynchronous until the register
- * is actually executed. It also has the option of returning a canceler,
- * which will be used by the runtime to cancel the asynchronous effect if the fiber
- * executing the effect is interrupted.
- *
- * If the register function returns a value synchronously, then the callback
- * function `Effect<A, E, R> => void` must not be called. Otherwise the callback
- * function must be called at most once.
- *
- * The `FiberId` of the fiber that may complete the async callback may be
- * provided to allow for better diagnostics.
- *
- * @since 2.0.0
- * @category constructors
- */
-export const asyncEither: <A, E = never, R = never>(
-  register: (callback: (effect: Effect<A, E, R>) => void) => Either.Either<Effect<void, never, R>, Effect<A, E, R>>,
-  blockingOn?: FiberId.FiberId
-) => Effect<A, E, R> = core.asyncEither
+export const asyncEffect: <A, E, R, R3, E2, R2>(
+  register: (callback: (_: Effect<A, E, R>) => void) => Effect<Effect<void, never, R3> | void, E2, R2>
+) => Effect<A, E | E2, R | R2 | R3> = _runtime.asyncEffect
 
 /**
  * @since 2.0.0

packages/effect/src/Stream.ts

@@ -256,15 +256,18 @@ export const as: {
 } = internal.as
 
 const _async: <A, E = never, R = never>(
-  register: (emit: Emit.Emit<R, E, A, void>) => void,
+  register: (emit: Emit.Emit<R, E, A, void>) => Effect.Effect<void, never, R> | void,
   outputBuffer?: number
 ) => Stream<A, E, R> = internal._async
 
 export {
   /**
    * Creates a stream from an asynchronous callback that can be called multiple
-   * times. The optionality of the error type `E` can be used to signal the end
-   * of the stream, by setting it to `None`.
+   * times. The optionality of the error type `E` in `Emit` can be used to
+   * signal the end of the stream by setting it to `None`.
+   *
+   * The registration function can optionally return an `Effect`, which will be
+   * executed if the `Fiber` executing this Effect is interrupted.
    *
    * @since 2.0.0
    * @category constructors
@@ -286,34 +289,6 @@ export const asyncEffect: <A, E = never, R = never>(
   outputBuffer?: number
 ) => Stream<A, E, R> = internal.asyncEffect
 
-/**
- * Creates a stream from an asynchronous callback that can be called multiple
- * times. The registration of the callback returns either a canceler or
- * synchronously returns a stream. The optionality of the error type `E` can
- * be used to signal the end of the stream, by setting it to `None`.
- *
- * @since 2.0.0
- * @category constructors
- */
-export const asyncInterrupt: <A, E = never, R = never>(
-  register: (emit: Emit.Emit<R, E, A, void>) => Either.Either<Effect.Effect<unknown, never, R>, Stream<A, E, R>>,
-  outputBuffer?: number
-) => Stream<A, E, R> = internal.asyncInterrupt
-
-/**
- * Creates a stream from an asynchronous callback that can be called multiple
- * times. The registration of the callback can possibly return the stream
- * synchronously. The optionality of the error type `E` can be used to signal
- * the end of the stream, by setting it to `None`.
- *
- * @since 2.0.0
- * @category constructors
- */
-export const asyncOption: <A, E = never, R = never>(
-  register: (emit: Emit.Emit<R, E, A, void>) => Option.Option<Stream<A, E, R>>,
-  outputBuffer?: number
-) => Stream<A, E, R> = internal.asyncOption
-
 /**
  * Creates a stream from an asynchronous callback that can be called multiple
  * times. The registration of the callback itself returns an a scoped

packages/effect/src/internal/clock.ts

@@ -2,7 +2,6 @@ import type * as Clock from "../Clock.js"
 import * as Context from "../Context.js"
 import * as Duration from "../Duration.js"
 import type * as Effect from "../Effect.js"
-import * as Either from "../Either.js"
 import { constFalse } from "../Function.js"
 import * as core from "./core.js"
 
@@ -85,9 +84,9 @@ class ClockImpl implements Clock.Clock {
   }
 
   sleep(duration: Duration.Duration): Effect.Effect<void> {
-    return core.asyncEither<void>((cb) => {
-      const canceler = globalClockScheduler.unsafeSchedule(() => cb(core.unit), duration)
-      return Either.left(core.asUnit(core.sync(canceler)))
+    return core.async<void>((resume) => {
+      const canceler = globalClockScheduler.unsafeSchedule(() => resume(core.unit), duration)
+      return core.asUnit(core.sync(canceler))
     })
   }
 }

packages/effect/src/internal/core-effect.ts

@@ -4,9 +4,8 @@ import * as Clock from "../Clock.js"
 import * as Context from "../Context.js"
 import * as Duration from "../Duration.js"
 import type * as Effect from "../Effect.js"
-import * as Either from "../Either.js"
 import type * as Fiber from "../Fiber.js"
-import * as FiberId from "../FiberId.js"
+import type * as FiberId from "../FiberId.js"
 import type * as FiberRef from "../FiberRef.js"
 import * as FiberRefs from "../FiberRefs.js"
 import type * as FiberRefsPatch from "../FiberRefsPatch.js"
@@ -75,26 +74,6 @@ export const asSome = <A, E, R>(self: Effect.Effect<A, E, R>): Effect.Effect<Opt
 export const asSomeError = <A, E, R>(self: Effect.Effect<A, E, R>): Effect.Effect<A, Option.Option<E>, R> =>
   core.mapError(self, Option.some)
 
-/* @internal */
-export const asyncOption = <A, E = never, R = never>(
-  register: (callback: (_: Effect.Effect<A, E, R>) => void) => Option.Option<Effect.Effect<A, E, R>>,
-  blockingOn: FiberId.FiberId = FiberId.none
-): Effect.Effect<A, E, R> =>
-  core.asyncEither(
-    (cb) => {
-      const option = register(cb)
-      switch (option._tag) {
-        case "None": {
-          return Either.left(core.unit)
-        }
-        case "Some": {
-          return Either.right(option.value)
-        }
-      }
-    },
-    blockingOn
-  )
-
 /* @internal */
 export const try_: {
   <A, E>(options: {

packages/effect/src/internal/core.ts

@@ -480,22 +480,6 @@ export const async = <A, E = never, R = never>(
       effect
   })
 
-/* @internal */
-export const asyncEither = <A, E = never, R = never>(
-  register: (
-    resume: (effect: Effect.Effect<A, E, R>) => void
-  ) => Either.Either<Effect.Effect<void, never, R>, Effect.Effect<A, E, R>>,
-  blockingOn: FiberId.FiberId = FiberId.none
-): Effect.Effect<A, E, R> =>
-  async<A, E, R>((resume) => {
-    const result = register(resume)
-    if (Either.isRight(result)) {
-      resume(result.right)
-    } else {
-      return result.left
-    }
-  }, blockingOn)
-
 /* @internal */
 export const catchAllCause = dual<
   <E, A2, E2, R2>(
@@ -1462,11 +1446,11 @@ export const zipWith: {
 ): Effect.Effect<B, E | E2, R | R2> => flatMap(self, (a) => map(that, (b) => f(a, b))))
 
 /* @internal */
-export const never: Effect.Effect<never> = asyncEither<never>(() => {
+export const never: Effect.Effect<never> = async<never>(() => {
   const interval = setInterval(() => {
     //
   }, 2 ** 31 - 1)
-  return Either.left(sync(() => clearInterval(interval)))
+  return sync(() => clearInterval(interval))
 })
 
 // -----------------------------------------------------------------------------
@@ -2844,18 +2828,18 @@ export const deferredMakeAs = <A, E = never>(fiberId: FiberId.FiberId): Effect.E
 
 /* @internal */
 export const deferredAwait = <A, E>(self: Deferred.Deferred<A, E>): Effect.Effect<A, E> =>
-  asyncEither<A, E>((k) => {
+  async<A, E>((resume) => {
     const state = MutableRef.get(self.state)
     switch (state._tag) {
       case DeferredOpCodes.OP_STATE_DONE: {
-        return Either.right(state.effect)
+        return resume(state.effect)
       }
       case DeferredOpCodes.OP_STATE_PENDING: {
         pipe(
           self.state,
-          MutableRef.set(deferred.pending([k, ...state.joiners]))
+          MutableRef.set(deferred.pending([resume, ...state.joiners]))
         )
-        return Either.left(deferredInterruptJoiner(self, k))
+        return deferredInterruptJoiner(self, resume)
       }
     }
   }, self.blockingOn)

packages/effect/src/internal/effect/circular.ts

@@ -2,7 +2,6 @@ import type * as Cause from "../../Cause.js"
 import type * as Deferred from "../../Deferred.js"
 import * as Duration from "../../Duration.js"
 import type * as Effect from "../../Effect.js"
-import * as Either from "../../Either.js"
 import * as Equal from "../../Equal.js"
 import type { Equivalence } from "../../Equivalence.js"
 import * as Exit from "../../Exit.js"
@@ -45,7 +44,7 @@ class Semaphore {
   }
 
   readonly take = (n: number): Effect.Effect<number> =>
-    core.asyncEither<number>((resume) => {
+    core.async<number>((resume) => {
       if (this.free < n) {
         const observer = () => {
           if (this.free < n) {
@@ -56,12 +55,12 @@ class Semaphore {
           resume(core.succeed(n))
         }
         this.waiters.add(observer)
-        return Either.left(core.sync(() => {
+        return core.sync(() => {
           this.waiters.delete(observer)
-        }))
+        })
       }
       this.taken += n
-      return Either.right(core.succeed(n))
+      return resume(core.succeed(n))
     })
 
   readonly updateTaken = (f: (n: number) => number): Effect.Effect<number> =>

packages/effect/src/internal/runtime.ts

@@ -22,6 +22,7 @@ import * as core from "./core.js"
 import * as executionStrategy from "./executionStrategy.js"
 import * as FiberRuntime from "./fiberRuntime.js"
 import * as fiberScope from "./fiberScope.js"
+import { internalize } from "./internalize.js"
 import * as OpCodes from "./opCodes/effect.js"
 import * as runtimeFlags from "./runtimeFlags.js"
 import * as _supervisor from "./supervisor.js"
@@ -450,22 +451,34 @@ export const unsafeRunSyncExitEffect = unsafeRunSyncExit(defaultRuntime)
 // circular with Effect
 
 /** @internal */
-export const asyncEffect = <A, E, R, X, E2, R2>(
-  register: (callback: (_: Effect.Effect<A, E, R>) => void) => Effect.Effect<X, E2, R2>
-): Effect.Effect<A, E | E2, R | R2> =>
-  core.flatMap(
-    core.deferredMake<A, E | E2>(),
-    (deferred) =>
-      core.flatMap(runtime<R | R2>(), (runtime) =>
-        core.uninterruptibleMask((restore) =>
-          core.zipRight(
-            FiberRuntime.fork(restore(
-              core.catchAllCause(
-                register((cb) => unsafeRunCallback(runtime)(core.intoDeferred(cb, deferred))),
-                (cause) => core.deferredFailCause(deferred, cause)
-              )
-            )),
-            restore(core.deferredAwait(deferred))
-          )
-        ))
-  )
+export const asyncEffect = <A, E, R, R3, E2, R2>(
+  register: (
+    callback: (_: Effect.Effect<A, E, R>) => void
+  ) => Effect.Effect<Effect.Effect<void, never, R3> | void, E2, R2>
+): Effect.Effect<A, E | E2, R | R2 | R3> =>
+  core.suspend(() => {
+    internalize(register)
+    let cleanup: Effect.Effect<void, never, R3> | void = undefined
+    return core.flatMap(
+      core.deferredMake<A, E | E2>(),
+      (deferred) =>
+        core.flatMap(runtime<R | R2 | R3>(), (runtime) =>
+          core.uninterruptibleMask((restore) =>
+            core.zipRight(
+              FiberRuntime.fork(restore(
+                core.matchCauseEffect(
+                  register((cb) => unsafeRunCallback(runtime)(core.intoDeferred(cb, deferred))),
+                  {
+                    onFailure: (cause) => core.deferredFailCause(deferred, cause),
+                    onSuccess: (cleanup_) => {
+                      cleanup = cleanup_
+                      return core.unit
+                    }
+                  }
+                )
+              )),
+              restore(core.onInterrupt(core.deferredAwait(deferred), () => cleanup ?? core.unit))
+            )
+          ))
+    )
+  })

packages/effect/src/internal/stream.ts

@@ -48,7 +48,7 @@ import * as pull from "./stream/pull.js"
 import * as SinkEndReason from "./stream/sinkEndReason.js"
 import * as ZipAllState from "./stream/zipAllState.js"
 import * as ZipChunksState from "./stream/zipChunksState.js"
-import * as _take from "./take.js"
+import * as InternalTake from "./take.js"
 
 /** @internal */
 const StreamSymbolKey = "effect/Stream"
@@ -461,13 +461,57 @@ export const as = dual<
 
 /** @internal */
 export const _async = <A, E = never, R = never>(
-  register: (emit: Emit.Emit<R, E, A, void>) => void,
+  register: (
+    emit: Emit.Emit<R, E, A, void>
+  ) => Effect.Effect<void, never, R> | void,
   outputBuffer = 16
 ): Stream.Stream<A, E, R> =>
-  asyncOption((cb) => {
-    register(cb)
-    return Option.none()
-  }, outputBuffer)
+  Effect.acquireRelease(
+    Queue.bounded<Take.Take<A, E>>(outputBuffer),
+    (queue) => Queue.shutdown(queue)
+  ).pipe(
+    Effect.flatMap((output) =>
+      Effect.runtime<R>().pipe(
+        Effect.flatMap((runtime) =>
+          Effect.sync(() => {
+            const runPromiseExit = Runtime.runPromiseExit(runtime)
+            const canceler = register(emit.make<R, E, A, void>((resume) =>
+              InternalTake.fromPull(resume).pipe(
+                Effect.flatMap((take) => Queue.offer(output, take)),
+                Effect.asUnit,
+                runPromiseExit
+              ).then((exit) => {
+                if (Exit.isFailure(exit)) {
+                  if (!Cause.isInterrupted(exit.cause)) {
+                    throw Cause.squash(exit.cause)
+                  }
+                }
+              })
+            ))
+            return canceler
+          })
+        ),
+        Effect.map((value) => {
+          const loop: Channel.Channel<Chunk.Chunk<A>, unknown, E, unknown, void, unknown> = Queue.take(output).pipe(
+            Effect.flatMap((take) => InternalTake.done(take)),
+            Effect.match({
+              onFailure: (maybeError) =>
+                core.fromEffect(Queue.shutdown(output)).pipe(
+                  channel.zipRight(Option.match(maybeError, {
+                    onNone: () => core.unit,
+                    onSome: (error) => core.fail(error)
+                  }))
+                ),
+              onSuccess: (chunk) => core.write(chunk).pipe(core.flatMap(() => loop))
+            }),
+            channel.unwrap
+          )
+          return fromChannel(loop).pipe(ensuring(value ?? Effect.unit))
+        })
+      )
+    ),
+    unwrapScoped
+  )
 
 /** @internal */
 export const asyncEffect = <A, E = never, R = never>(
@@ -487,7 +531,7 @@ export const asyncEffect = <A, E = never, R = never>(
             register(
               emit.make((k) =>
                 pipe(
-                  _take.fromPull(k),
+                  InternalTake.fromPull(k),
                   Effect.flatMap((take) => Queue.offer(output, take)),
                   Effect.asUnit,
                   Runtime.runPromiseExit(runtime)
@@ -503,7 +547,7 @@ export const asyncEffect = <A, E = never, R = never>(
             Effect.map(() => {
               const loop: Channel.Channel<Chunk.Chunk<A>, unknown, E, unknown, void, unknown> = pipe(
                 Queue.take(output),
-                Effect.flatMap(_take.done),
+                Effect.flatMap(InternalTake.done),
                 Effect.match({
                   onFailure: (maybeError) =>
                     pipe(
@@ -524,84 +568,6 @@ export const asyncEffect = <A, E = never, R = never>(
     fromChannel
   )
 
-/** @internal */
-export const asyncInterrupt = <A, E = never, R = never>(
-  register: (
-    emit: Emit.Emit<R, E, A, void>
-  ) => Either.Either<Effect.Effect<unknown, never, R>, Stream.Stream<A, E, R>>,
-  outputBuffer = 16
-): Stream.Stream<A, E, R> =>
-  pipe(
-    Effect.acquireRelease(
-      Queue.bounded<Take.Take<A, E>>(outputBuffer),
-      (queue) => Queue.shutdown(queue)
-    ),
-    Effect.flatMap((output) =>
-      pipe(
-        Effect.runtime<R>(),
-        Effect.flatMap((runtime) =>
-          pipe(
-            Effect.sync(() =>
-              register(
-                emit.make((k) =>
-                  pipe(
-                    _take.fromPull(k),
-                    Effect.flatMap((take) => Queue.offer(output, take)),
-                    Effect.asUnit,
-                    Runtime.runPromiseExit(runtime)
-                  ).then((exit) => {
-                    if (Exit.isFailure(exit)) {
-                      if (!Cause.isInterrupted(exit.cause)) {
-                        throw Cause.squash(exit.cause)
-                      }
-                    }
-                  })
-                )
-              )
-            ),
-            Effect.map(Either.match({
-              onLeft: (canceler) => {
-                const loop: Channel.Channel<Chunk.Chunk<A>, unknown, E, unknown, void, unknown> = pipe(
-                  Queue.take(output),
-                  Effect.flatMap(_take.done),
-                  Effect.match({
-                    onFailure: (maybeError) =>
-                      channel.zipRight(
-                        core.fromEffect(Queue.shutdown(output)),
-                        Option.match(maybeError, {
-                          onNone: () => core.unit,
-                          onSome: core.fail
-                        })
-                      ),
-                    onSuccess: (chunk) => pipe(core.write(chunk), core.flatMap(() => loop))
-                  }),
-                  channel.unwrap
-                )
-                return pipe(fromChannel(loop), ensuring(canceler))
-              },
-              onRight: (stream) => unwrap(pipe(Queue.shutdown(output), Effect.as(stream)))
-            }))
-          )
-        )
-      )
-    ),
-    unwrapScoped
-  )
-
-/** @internal */
-export const asyncOption = <A, E = never, R = never>(
-  register: (emit: Emit.Emit<R, E, A, void>) => Option.Option<Stream.Stream<A, E, R>>,
-  outputBuffer = 16
-): Stream.Stream<A, E, R> =>
-  asyncInterrupt(
-    (emit) =>
-      Option.match(register(emit), {
-        onNone: () => Either.left(Effect.unit),
-        onSome: Either.right
-      }),
-    outputBuffer
-  )
-
 /** @internal */
 export const asyncScoped = <A, E = never, R = never>(
   register: (emit: Emit.Emit<R, E, A, void>) => Effect.Effect<unknown, E, R | Scope.Scope>,
@@ -620,7 +586,7 @@ export const asyncScoped = <A, E = never, R = never>(
             register(
               emit.make((k) =>
                 pipe(
-                  _take.fromPull(k),
+                  InternalTake.fromPull(k),
                   Effect.flatMap((take) => Queue.offer(output, take)),
                   Effect.asUnit,
                   Runtime.runPromiseExit(runtime)
@@ -642,7 +608,7 @@ export const asyncScoped = <A, E = never, R = never>(
                     pull.end() :
                     pipe(
                       Queue.take(output),
-                      Effect.flatMap(_take.done),
+                      Effect.flatMap(InternalTake.done),
                       Effect.onError(() =>
                         pipe(
                           Ref.set(ref, true),
@@ -884,7 +850,7 @@ export const bufferChunks = dual<
       Effect.map(queue, (queue) => {
         const process: Channel.Channel<Chunk.Chunk<A>, unknown, E, unknown, void, unknown> = pipe(
           core.fromEffect(Queue.take(queue)),
-          core.flatMap(_take.match({
+          core.flatMap(InternalTake.match({
             onEnd: () => core.unit,
             onFailure: core.failCause,
             onSuccess: (value) => pipe(core.write(value), core.flatMap(() => process))
@@ -947,7 +913,7 @@ const bufferUnbounded = <A, E, R>(self: Stream.Stream<A, E, R>): Stream.Stream<A
       Effect.map(queue, (queue) => {
         const process: Channel.Channel<Chunk.Chunk<A>, unknown, E, unknown, void, unknown> = pipe(
           core.fromEffect(Queue.take(queue)),
-          core.flatMap(_take.match({
+          core.flatMap(InternalTake.match({
             onEnd: () => core.unit,
             onFailure: core.failCause,
             onSuccess: (value) => core.flatMap(core.write(value), () => process)
@@ -989,16 +955,16 @@ const bufferSignal = <A, E, R>(
           Effect.flatMap(
             (deferred) =>
               pipe(
-                Queue.offer(queue, [_take.chunk(input), deferred] as const),
+                Queue.offer(queue, [InternalTake.chunk(input), deferred] as const),
                 Effect.flatMap((added) => pipe(Ref.set(ref, deferred), Effect.when(() => added)))
               )
           ),
           Effect.asUnit,
           core.fromEffect,
           core.flatMap(() => producer(queue, ref))
         ),
-      onFailure: (error) => terminate(_take.failCause(error)),
-      onDone: () => terminate(_take.end)
+      onFailure: (error) => terminate(InternalTake.failCause(error)),
+      onDone: () => terminate(InternalTake.end)
     })
   }
   const consumer = (
@@ -1009,7 +975,7 @@ const bufferSignal = <A, E, R>(
       core.flatMap(([take, deferred]) =>
         channel.zipRight(
           core.fromEffect(Deferred.succeed(deferred, void 0)),
-          _take.match(take, {
+          InternalTake.match(take, {
             onEnd: () => core.unit,
             onFailure: core.failCause,
             onSuccess: (value) => pipe(core.write(value), core.flatMap(() => process))
@@ -1469,19 +1435,19 @@ export const combineChunks = dual<
           core.flatMap(
             core.fromEffect(pipe(
               handoff,
-              Handoff.offer<Take.Take<Elem, Err>>(_take.chunk(input))
+              Handoff.offer<Take.Take<Elem, Err>>(InternalTake.chunk(input))
             )),
             () => producer(handoff, latch)
           ),
         onFailure: (cause) =>
           core.fromEffect(
             Handoff.offer<Take.Take<Elem, Err>>(
               handoff,
-              _take.failCause(cause)
+              InternalTake.failCause(cause)
             )
           ),
         onDone: (): Channel.Channel<never, Chunk.Chunk<Elem>, never, Err, unknown, unknown, R> =>
-          core.fromEffect(Handoff.offer<Take.Take<Elem, Err>>(handoff, _take.end))
+          core.fromEffect(Handoff.offer<Take.Take<Elem, Err>>(handoff, InternalTake.end))
       })
     )
   return new StreamImpl(
@@ -1515,7 +1481,7 @@ export const combineChunks = dual<
           Effect.zipRight(
             pipe(
               Handoff.take(left),
-              Effect.flatMap(_take.done)
+              Effect.flatMap(InternalTake.done)
             )
           )
         )
@@ -1525,7 +1491,7 @@ export const combineChunks = dual<
           Effect.zipRight(
             pipe(
               Handoff.take(right),
-              Effect.flatMap(_take.done)
+              Effect.flatMap(InternalTake.done)
             )
           )
         )
@@ -3388,20 +3354,20 @@ export const interleaveWith = dual<
         onInput: (value: A | A2) =>
           core.flatMap(
             core.fromEffect(
-              Handoff.offer<Take.Take<A | A2, E | E2 | E3>>(handoff, _take.of(value))
+              Handoff.offer<Take.Take<A | A2, E | E2 | E3>>(handoff, InternalTake.of(value))
             ),
             () => producer(handoff)
           ),
         onFailure: (cause) =>
           core.fromEffect(
             Handoff.offer<Take.Take<A | A2, E | E2 | E3>>(
               handoff,
-              _take.failCause(cause)
+              InternalTake.failCause(cause)
             )
           ),
         onDone: () =>
           core.fromEffect(
-            Handoff.offer<Take.Take<A | A2, E | E2 | E3>>(handoff, _take.end)
+            Handoff.offer<Take.Take<A | A2, E | E2 | E3>>(handoff, InternalTake.end)
           )
       })
     return new StreamImpl(
@@ -3437,7 +3403,7 @@ export const interleaveWith = dual<
                   if (bool && !leftDone) {
                     return pipe(
                       core.fromEffect(Handoff.take(left)),
-                      core.flatMap(_take.match({
+                      core.flatMap(InternalTake.match({
                         onEnd: () => rightDone ? core.unit : process(true, rightDone),
                         onFailure: core.failCause,
                         onSuccess: (chunk) => pipe(core.write(chunk), core.flatMap(() => process(leftDone, rightDone)))
@@ -3447,7 +3413,7 @@ export const interleaveWith = dual<
                   if (!bool && !rightDone) {
                     return pipe(
                       core.fromEffect(Handoff.take(right)),
-                      core.flatMap(_take.match({
+                      core.flatMap(InternalTake.match({
                         onEnd: () => leftDone ? core.unit : process(leftDone, true),
                         onFailure: core.failCause,
                         onSuccess: (chunk) => pipe(core.write(chunk), core.flatMap(() => process(leftDone, rightDone)))
@@ -5450,9 +5416,9 @@ export const runIntoQueueScoped = dual<
 ): Effect.Effect<void, never, Scope.Scope | R> => {
   const writer: Channel.Channel<Take.Take<A, E>, Chunk.Chunk<A>, never, E, unknown, unknown, R> = core
     .readWithCause({
-      onInput: (input: Chunk.Chunk<A>) => core.flatMap(core.write(_take.chunk(input)), () => writer),
-      onFailure: (cause) => core.write(_take.failCause(cause)),
-      onDone: () => core.write(_take.end)
+      onInput: (input: Chunk.Chunk<A>) => core.flatMap(core.write(InternalTake.chunk(input)), () => writer),
+      onFailure: (cause) => core.write(InternalTake.failCause(cause)),
+      onDone: () => core.write(InternalTake.end)
     })
   return pipe(
     core.pipeTo(toChannel(self), writer),
@@ -6229,23 +6195,23 @@ export const tapSink = dual<
           .readWithCause({
             onInput: (chunk: Chunk.Chunk<A>) =>
               pipe(
-                core.fromEffect(Queue.offer(queue, _take.chunk(chunk))),
+                core.fromEffect(Queue.offer(queue, InternalTake.chunk(chunk))),
                 core.foldCauseChannel({
                   onFailure: () => core.flatMap(core.write(chunk), () => channel.identityChannel()),
                   onSuccess: () => core.flatMap(core.write(chunk), () => loop)
                 })
               ) as Channel.Channel<Chunk.Chunk<A>, Chunk.Chunk<A>, E | E2, E, unknown, unknown, R2>,
             onFailure: (cause: Cause.Cause<E | E2>) =>
               pipe(
-                core.fromEffect(Queue.offer(queue, _take.failCause(cause))),
+                core.fromEffect(Queue.offer(queue, InternalTake.failCause(cause))),
                 core.foldCauseChannel({
                   onFailure: () => core.failCause(cause),
                   onSuccess: () => core.failCause(cause)
                 })
               ),
             onDone: () =>
               pipe(
-                core.fromEffect(Queue.offer(queue, _take.end)),
+                core.fromEffect(Queue.offer(queue, InternalTake.end)),
                 core.foldCauseChannel({
                   onFailure: () => core.unit,
                   onSuccess: () => core.unit
@@ -6256,7 +6222,7 @@ export const tapSink = dual<
           new StreamImpl(pipe(
             core.pipeTo(toChannel(self), loop),
             channel.ensuring(Effect.zipRight(
-              Effect.forkDaemon(Queue.offer(queue, _take.end)),
+              Effect.forkDaemon(Queue.offer(queue, InternalTake.end)),
               Deferred.await(deferred)
             ))
           )),

packages/effect/test/Effect/async.test.ts

@@ -22,8 +22,8 @@ describe("Effect", () => {
     }))
   it.effect("simple asyncEffect must return", () =>
     Effect.gen(function*($) {
-      const result = yield* $(Effect.asyncEffect<unknown, unknown, never, void, never, never>((cb) => {
-        return Effect.succeed(cb(Effect.succeed(42)))
+      const result = yield* $(Effect.asyncEffect<number, never, never, never, never, never>((resume) => {
+        return Effect.succeed(resume(Effect.succeed(42)))
       }))
       assert.strictEqual(result, 42)
     }))
@@ -95,18 +95,18 @@ describe("Effect", () => {
       assert.deepStrictEqual(unexpected, Chunk.empty())
       assert.deepStrictEqual(result, Option.none()) // the timeout should happen
     }))
-  it.live("asyncMaybe should not resume fiber twice after synchronous result", () =>
+  it.live("async should not resume fiber twice after synchronous result", () =>
     Effect.gen(function*($) {
       const step = yield* $(Deferred.make<void>())
       const unexpectedPlace = yield* $(Ref.make(Chunk.empty<number>()))
       const runtime = yield* $(Effect.runtime<never>())
       const fiber = yield* $(
-        Effect.asyncOption<void, never, never>((cb) => {
+        Effect.async<void, never, never>((resume) => {
           Runtime.runCallback(runtime)(pipe(
             Deferred.await(step),
-            Effect.zipRight(Effect.sync(() => cb(Ref.update(unexpectedPlace, Chunk.prepend(1)))))
+            Effect.zipRight(Effect.sync(() => resume(Ref.update(unexpectedPlace, Chunk.prepend(1)))))
           ))
-          return Option.some(Effect.unit)
+          return Effect.unit
         }),
         Effect.flatMap(() =>
           Effect.async<void, never, never>(() => {

packages/effect/test/Effect/concurrency.test.ts

@@ -49,7 +49,7 @@ describe("Effect", () => {
       const release = yield* $(Deferred.make<number>())
       const acquire = yield* $(Deferred.make<void>())
       const fiber = yield* $(
-        Effect.asyncEffect<unknown, unknown, never, unknown, unknown, never>((_) =>
+        Effect.asyncEffect<void, never, never, never, never, never>((_) =>
           // This will never complete because the callback is never invoked
           Effect.acquireUseRelease(
             Deferred.succeed(acquire, void 0),

packages/effect/test/Effect/interruption.test.ts

@@ -546,11 +546,11 @@ describe("Effect", () => {
   it.effect("async cancelation", () =>
     Effect.gen(function*($) {
       const ref = MutableRef.make(0)
-      const effect = Effect.asyncEither(() => {
+      const effect = Effect.async(() => {
         pipe(ref, MutableRef.set(MutableRef.get(ref) + 1))
-        return Either.left(Effect.sync(() => {
+        return Effect.sync(() => {
           pipe(ref, MutableRef.set(MutableRef.get(ref) - 1))
-        }))
+        })
       })
       yield* $(Effect.unit, Effect.race(effect))
       const result = MutableRef.get(ref)