Function _async

packages/effect/src/internal/stream.ts:481–535

(
  register: (
    emit: Emit.Emit<R, E, A, void>
  ) => Effect.Effect<void, never, R> | void,
  bufferSize?: number | "unbounded" | {
    readonly bufferSize?: number | undefined
    readonly strategy?: "dropping" | "sliding" | "suspend" | undefined
  } | undefined
)

/** @internal */
export const _async = <A, E = never, R = never>(
  register: (
    emit: Emit.Emit<R, E, A, void>
  ) => Effect.Effect<void, never, R> | void,
  bufferSize?: number | "unbounded" | {
    readonly bufferSize?: number | undefined
    readonly strategy?: "dropping" | "sliding" | "suspend" | undefined
  } | undefined
): Stream.Stream<A, E, R> =>
  Effect.acquireRelease(
    queueFromBufferOptions<A, E>(bufferSize),
    (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.asVoid,
                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.void,
                    onSome: (error) => core.fail(error)
                  }))
                ),
              onSuccess: (chunk) => core.write(chunk).pipe(core.flatMap(() => loop))
            }),
            channel.unwrap
          )
          return fromChannel(loop).pipe(ensuring(value ?? Effect.void))
        })
      )
    ),
    unwrapScoped
  )

/** @internal */
export const asyncEffect = <A, E = never, R = never>(