Function flattenExitOption

packages/effect/src/internal/stream.ts:2912–2960

(
  self: Stream.Stream<Exit.Exit<A, Option.Option<E2>>, E, R>
)

/** @internal */
export const flattenExitOption = <A, E2, E, 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>>>,
    E | E2,
    E,
    unknown,
    unknown,
    R
  > = core.readWithCause({
    onInput: (chunk: Chunk.Chunk<Exit.Exit<A, Option.Option<E2>>>) => processChunk(chunk, process),
    onFailure: (cause) => core.failCause<E | E2>(cause),
    onDone: () => core.void
  })
  return new StreamImpl(pipe(toChannel(self), core.pipeTo(process)))
}

/** @internal */
export const flattenIterables = <A, E, R>(self: Stream.Stream<Iterable<A>, E, R>): Stream.Stream<A, E, R> =>