Function dropUntilEffectReader

packages/effect/src/internal/sink.ts:582–603

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

/** @internal */
const dropUntilEffectReader = <In, R, E>(
  predicate: (input: In) => Effect.Effect<boolean, E, R>
): Channel.Channel<Chunk.Chunk<In>, Chunk.Chunk<In>, E, E, unknown, unknown, R> =>
  core.readWith({
    onInput: (input: Chunk.Chunk<In>) =>
      pipe(
        input,
        Effect.dropUntil(predicate),
        Effect.map((leftover) => {
          const more = leftover.length === 0
          return more ?
            dropUntilEffectReader(predicate) :
            pipe(
              core.write(Chunk.unsafeFromArray(leftover)),
              channel.zipRight(channel.identityChannel<Chunk.Chunk<In>, E, unknown>())
            )
        }),
        channel.unwrap
      ),
    onFailure: core.fail,
    onDone: () => core.void
  })

/** @internal */
export const dropWhile = <In>(predicate: Predicate<In>): Sink.Sink<unknown, In, In> =>