Function asyncPush

packages/effect/src/internal/stream.ts:613–638

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

/** @internal */
export const asyncPush = <A, E = never, R = never>(
  register: (emit: Emit.EmitOpsPush<E, A>) => Effect.Effect<unknown, E, R | Scope.Scope>,
  options?: {
    readonly bufferSize: "unbounded"
  } | {
    readonly bufferSize?: number | undefined
    readonly strategy?: "dropping" | "sliding" | undefined
  } | undefined
): Stream.Stream<A, E, Exclude<R, Scope.Scope>> =>
  Effect.acquireRelease(
    queueFromBufferOptionsPush<A, E>(options),
    Queue.shutdown
  ).pipe(
    Effect.tap((queue) =>
      FiberRef.getWith(FiberRef.currentScheduler, (scheduler) => register(emit.makePush(queue, scheduler)))
    ),
    Effect.map((queue) => {
      const loop: Channel.Channel<Chunk.Chunk<A>, unknown, E> = core.flatMap(Queue.take(queue), (item) =>
        Exit.isExit(item)
          ? Exit.isSuccess(item) ? core.void : core.failCause(item.cause)
          : channel.zipRight(core.write(Chunk.unsafeFromArray(item)), loop))
      return loop
    }),
    channel.unwrapScoped,
    fromChannel
  )

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