Function asyncEffect

packages/effect/src/internal/stream.ts:538–593

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

/** @internal */
export const asyncEffect = <A, E = never, R = never>(
  register: (emit: Emit.Emit<R, E, A, void>) => Effect.Effect<unknown, E, R>,
  bufferSize?: number | "unbounded" | {
    readonly bufferSize?: number | undefined
    readonly strategy?: "dropping" | "sliding" | "suspend" | undefined
  } | undefined
): Stream.Stream<A, E, R> =>
  pipe(
    Effect.acquireRelease(
      queueFromBufferOptions<A, E>(bufferSize),
      (queue) => Queue.shutdown(queue)
    ),
    Effect.flatMap((output) =>
      pipe(
        Effect.runtime<R>(),
        Effect.flatMap((runtime) =>
          pipe(
            register(
              emit.make((k) =>
                pipe(
                  InternalTake.fromPull(k),
                  Effect.flatMap((take) => Queue.offer(output, take)),
                  Effect.asVoid,
                  Runtime.runPromiseExit(runtime)
                ).then((exit) => {
                  if (Exit.isFailure(exit)) {
                    if (!Cause.isInterrupted(exit.cause)) {
                      throw Cause.squash(exit.cause)
                    }
                  }
                })
              )
            ),
            Effect.map(() => {
              const loop: Channel.Channel<Chunk.Chunk<A>, unknown, E, unknown, void, unknown> = pipe(
                Queue.take(output),
                Effect.flatMap(InternalTake.done),
                Effect.match({
                  onFailure: (maybeError) =>
                    pipe(
                      core.fromEffect(Queue.shutdown(output)),
                      channel.zipRight(Option.match(maybeError, { onNone: () => core.void, onSome: core.fail }))
                    ),
                  onSuccess: (chunk) => pipe(core.write(chunk), core.flatMap(() => loop))
                }),
                channel.unwrap
              )
              return loop
            })
          )
        )
      )
    ),
    channel.unwrapScoped,
    fromChannel
  )

const queueFromBufferOptionsPush = <A, E>(