Function collectAllWhileEffectReader

packages/effect/src/internal/sink.ts:237–255

(
  predicate: (input: In) => Effect.Effect<boolean, E, R>,
  done: Chunk.Chunk<In>
)

/** @internal */
const collectAllWhileEffectReader = <In, R, E>(
  predicate: (input: In) => Effect.Effect<boolean, E, R>,
  done: Chunk.Chunk<In>
): Channel.Channel<Chunk.Chunk<In>, Chunk.Chunk<In>, E, never, Chunk.Chunk<In>, unknown, R> =>
  core.readWith({
    onInput: (input: Chunk.Chunk<In>) =>
      pipe(
        core.fromEffect(pipe(input, Effect.takeWhile(predicate), Effect.map(Chunk.unsafeFromArray))),
        core.flatMap((collected) => {
          const leftovers = pipe(input, Chunk.drop(collected.length))
          if (Chunk.isEmpty(leftovers)) {
            return collectAllWhileEffectReader(predicate, pipe(done, Chunk.appendAll(collected)))
          }
          return pipe(core.write(leftovers), channel.zipRight(core.succeed(pipe(done, Chunk.appendAll(collected)))))
        })
      ),
    onFailure: core.fail,
    onDone: () => core.succeed(done)
  })

/** @internal */
export const collectAllWhileWith: {