Function accumulator

packages/effect/src/internal/stream.ts:102–115

(
    s: Chunk.Chunk<A>
  )

/** @internal */
export const accumulateChunks = <A, E, R>(self: Stream.Stream<A, E, R>): Stream.Stream<A, E, R> => {
  const accumulator = (
    s: Chunk.Chunk<A>
  ): Channel.Channel<Chunk.Chunk<A>, Chunk.Chunk<A>, E, E, void, unknown> =>
    core.readWith({
      onInput: (input: Chunk.Chunk<A>) => {
        const next = Chunk.appendAll(s, input)
        return core.flatMap(
          core.write(next),
          () => accumulator(next)
        )
      },
      onFailure: core.fail,
      onDone: () => core.void
    })
  return new StreamImpl(core.pipeTo(toChannel(self), accumulator(Chunk.empty())))
}