Function fromPushPull

packages/effect/src/internal/sink.ts:1444–1474

(
  push: (
    option: Option.Option<Chunk.Chunk<In>>
  ) => Effect.Effect<void, readonly [Either.Either<Z, E>, Chunk.Chunk<L>], R>
)

  new SinkImpl(channel.unwrapScoped(pipe(push, Effect.map(fromPushPull))))

const fromPushPull = <In, Z, E, L, R>(
  push: (
    option: Option.Option<Chunk.Chunk<In>>
  ) => Effect.Effect<void, readonly [Either.Either<Z, E>, Chunk.Chunk<L>], R>
): Channel.Channel<Chunk.Chunk<L>, Chunk.Chunk<In>, E, never, Z, unknown, R> =>
  core.readWith({
    onInput: (input: Chunk.Chunk<In>) =>
      channel.foldChannel(core.fromEffect(push(Option.some(input))), {
        onFailure: ([either, leftovers]) =>
          Either.match(either, {
            onLeft: (error) => pipe(core.write(leftovers), channel.zipRight(core.fail(error))),
            onRight: (z) => pipe(core.write(leftovers), channel.zipRight(core.succeedNow(z)))
          }),
        onSuccess: () => fromPushPull(push)
      }),
    onFailure: core.fail,
    onDone: () =>
      channel.foldChannel(core.fromEffect(push(Option.none())), {
        onFailure: ([either, leftovers]) =>
          Either.match(either, {
            onLeft: (error) => pipe(core.write(leftovers), channel.zipRight(core.fail(error))),
            onRight: (z) => pipe(core.write(leftovers), channel.zipRight(core.succeedNow(z)))
          }),
        onSuccess: () =>
          core.fromEffect(
            Effect.dieMessage(
              "BUG: Sink.fromPush - please report an issue at https://github.com/Effect-TS/effect/issues"
            )
          )
      })
  })

/** @internal */
export const fromQueue = <In>(