Function processChunk

packages/effect/src/internal/stream.ts:2915–2945

(
    chunk: Chunk.Chunk<Exit.Exit<A, Option.Option<E2>>>,
    cont: Channel.Channel<Chunk.Chunk<A>, Chunk.Chunk<Exit.Exit<A, Option.Option<E2>>>, E | E2, E, unknown, unknown, R>
  )

  self: Stream.Stream<Exit.Exit<A, Option.Option<E2>>, E, R>
): Stream.Stream<A, E | E2, R> => {
  const processChunk = (
    chunk: Chunk.Chunk<Exit.Exit<A, Option.Option<E2>>>,
    cont: Channel.Channel<Chunk.Chunk<A>, Chunk.Chunk<Exit.Exit<A, Option.Option<E2>>>, E | E2, E, unknown, unknown, R>
  ) => {
    const [toEmit, rest] = pipe(chunk, Chunk.splitWhere((exit) => !Exit.isSuccess(exit)))
    const next = pipe(
      Chunk.head(rest),
      Option.match({
        onNone: () => cont,
        onSome: Exit.match({
          onFailure: (cause) =>
            Option.match(Cause.flipCauseOption(cause), {
              onNone: () => core.void,
              onSome: core.failCause
            }),
          onSuccess: () => core.void
        })
      })
    )
    return pipe(
      core.write(pipe(
        toEmit,
        Chunk.filterMap((exit) =>
          Exit.isSuccess(exit) ?
            Option.some(exit.value) :
            Option.none()
        )
      )),
      core.flatMap(() => next)
    )
  }
  const process: Channel.Channel<
    Chunk.Chunk<A>,
    Chunk.Chunk<Exit.Exit<A, Option.Option<E2>>>,