Function collectAllWhileWithLoop

packages/effect/src/internal/sink.ts:310–344

(
  self: Sink.Sink<Z, In, L, E, R>,
  leftoversRef: Ref.Ref<Chunk.Chunk<In>>,
  upstreamDoneRef: Ref.Ref<boolean>,
  currentResult: S,
  p: Predicate<Z>,
  f: (s: S, z: Z) => S
)

)

const collectAllWhileWithLoop = <Z, In, L extends In, E, R, S>(
  self: Sink.Sink<Z, In, L, E, R>,
  leftoversRef: Ref.Ref<Chunk.Chunk<In>>,
  upstreamDoneRef: Ref.Ref<boolean>,
  currentResult: S,
  p: Predicate<Z>,
  f: (s: S, z: Z) => S
): Channel.Channel<Chunk.Chunk<L>, Chunk.Chunk<In>, E, never, S, unknown, R> => {
  return pipe(
    toChannel(self),
    channel.doneCollect,
    channel.foldChannel({
      onFailure: core.fail,
      onSuccess: ([leftovers, doneValue]) =>
        p(doneValue)
          ? pipe(
            core.fromEffect(
              Ref.set(leftoversRef, Chunk.flatten(leftovers as Chunk.Chunk<Chunk.Chunk<In>>))
            ),
            core.flatMap(() =>
              pipe(
                core.fromEffect(Ref.get(upstreamDoneRef)),
                core.flatMap((upstreamDone) => {
                  const accumulatedResult = f(currentResult, doneValue)
                  return upstreamDone
                    ? pipe(core.write(Chunk.flatten(leftovers)), channel.as(accumulatedResult))
                    : collectAllWhileWithLoop(self, leftoversRef, upstreamDoneRef, accumulatedResult, p, f)
                })
              )
            )
          )
          : pipe(core.write(Chunk.flatten(leftovers)), channel.as(currentResult))
    })
  )
}

/** @internal */
export const collectLeftover = <A, In, L, E, R>(