Function takeUntil

repos/effect/packages/effect/src/Sink.ts:1716–1724

(predicate: Predicate<In>)

    const out = Arr.empty<Out>()
    let leftover: Arr.NonEmptyReadonlyArray<In> | undefined = undefined
    return upstream.pipe(
      Effect.flatMap((arr) => {
        let i = 0
        return Effect.whileLoop({
          while: () => i < arr.length,
          body: constant(Effect.flatMap(Effect.suspend(() => filter(arr[i++])), (result) => {
            if (Result.isFailure(result)) {
              if (i < arr.length) {
                leftover = arr.slice(i) as any
              }
              return Cause.done()
            }