Function flattenChunks

packages/effect/src/internal/stream.ts:2860–2872

(self: Stream.Stream<Chunk.Chunk<A>, E, R>)

/** @internal */
export const flattenChunks = <A, E, R>(self: Stream.Stream<Chunk.Chunk<A>, E, R>): Stream.Stream<A, E, R> => {
  const flatten: Channel.Channel<Chunk.Chunk<A>, Chunk.Chunk<Chunk.Chunk<A>>, E, E, unknown, unknown> = core
    .readWithCause({
      onInput: (chunks: Chunk.Chunk<Chunk.Chunk<A>>) =>
        core.flatMap(
          channel.writeChunk(chunks),
          () => flatten
        ),
      onFailure: core.failCause,
      onDone: () => core.void
    })
  return new StreamImpl(pipe(toChannel(self), core.pipeTo(flatten)))
}

/** @internal */
export const flattenEffect = dual<