Function producer

packages/effect/src/internal/stream.ts:1133–1170

(
    queue: Queue.Queue<readonly [Take.Take<A, E>, Deferred.Deferred<void>]>,
    ref: Ref.Ref<Deferred.Deferred<void>>
  )

  bufferChannel: Channel.Channel<Chunk.Chunk<A>, unknown, E, unknown, void, unknown, R>
): Channel.Channel<Chunk.Chunk<A>, unknown, E, unknown, void, unknown, R> => {
  const producer = (
    queue: Queue.Queue<readonly [Take.Take<A, E>, Deferred.Deferred<void>]>,
    ref: Ref.Ref<Deferred.Deferred<void>>
  ): Channel.Channel<never, Chunk.Chunk<A>, never, E, unknown, unknown, R> => {
    const terminate = (take: Take.Take<A, E>): Channel.Channel<never, Chunk.Chunk<A>, never, E, unknown, unknown, R> =>
      pipe(
        Ref.get(ref),
        Effect.tap(Deferred.await),
        Effect.zipRight(Deferred.make<void>()),
        Effect.flatMap((deferred) =>
          pipe(
            Queue.offer(queue, [take, deferred] as const),
            Effect.zipRight(Ref.set(ref, deferred)),
            Effect.zipRight(Deferred.await(deferred))
          )
        ),
        Effect.asVoid,
        core.fromEffect
      )
    return core.readWithCause({
      onInput: (input: Chunk.Chunk<A>) =>
        pipe(
          Deferred.make<void>(),
          Effect.flatMap(
            (deferred) =>
              pipe(
                Queue.offer(queue, [InternalTake.chunk(input), deferred] as const),
                Effect.flatMap((added) => pipe(Ref.set(ref, deferred), Effect.when(() => added)))
              )
          ),
          Effect.asVoid,
          core.fromEffect,
          core.flatMap(() => producer(queue, ref))
        ),
      onFailure: (error) => terminate(InternalTake.failCause(error)),
      onDone: () => terminate(InternalTake.end)
    })
  }
  const consumer = (
    queue: Queue.Queue<readonly [Take.Take<A, E>, Deferred.Deferred<void>]>
  ): Channel.Channel<Chunk.Chunk<A>, unknown, E, unknown, void, unknown, R> => {