Function consumer

packages/effect/src/internal/stream.ts:1171–1188

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

    })
  }
  const consumer = (
    queue: Queue.Queue<readonly [Take.Take<A, E>, Deferred.Deferred<void>]>
  ): Channel.Channel<Chunk.Chunk<A>, unknown, E, unknown, void, unknown, R> => {
    const process: Channel.Channel<Chunk.Chunk<A>, unknown, E, unknown, void, unknown> = pipe(
      core.fromEffect(Queue.take(queue)),
      core.flatMap(([take, deferred]) =>
        channel.zipRight(
          core.fromEffect(Deferred.succeed(deferred, void 0)),
          InternalTake.match(take, {
            onEnd: () => core.void,
            onFailure: core.failCause,
            onSuccess: (value) => pipe(core.write(value), core.flatMap(() => process))
          })
        )
      )
    )
    return process
  }
  return channel.unwrapScoped(
    pipe(
      scoped,