Hyperlinkv0.8.0-beta.28

Stream

Stream.takeUntilEffectconsteffect/Stream.ts:6473
<A, E2, R2>(
  predicate: (a: NoInfer<A>, n: number) => Effect.Effect<boolean, E2, R2>,
  options?: { readonly excludeLast?: boolean | undefined }
): <E, R>(self: Stream<A, E, R>) => Stream<A, E2 | E, R2 | R>
<A, E, R, E2, R2>(
  self: Stream<A, E, R>,
  predicate: (a: A, n: number) => Effect.Effect<boolean, E2, R2>,
  options?: { readonly excludeLast?: boolean | undefined }
): Stream<A, E | E2, R | R2>

Takes stream elements until an effectful predicate returns true.

When to use

Use when the stopping condition needs an Effect or service and predicate failure should fail the stream.

Example (Taking until an effectful predicate matches)

import { Console, Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const result = yield* Stream.range(1, 5).pipe(
    Stream.takeUntilEffect((n) => Effect.succeed(n % 3 === 0)),
    Stream.runCollect
  )
  yield* Console.log(result)
})

Effect.runPromise(program)
// Output: [ 1, 2, 3 ]
filtering
Source effect/Stream.ts:647338 lines
export const takeUntilEffect: {
  <A, E2, R2>(
    predicate: (a: NoInfer<A>, n: number) => Effect.Effect<boolean, E2, R2>,
    options?: {
      readonly excludeLast?: boolean | undefined
    }
  ): <E, R>(self: Stream<A, E, R>) => Stream<A, E2 | E, R2 | R>
  <A, E, R, E2, R2>(
    self: Stream<A, E, R>,
    predicate: (a: A, n: number) => Effect.Effect<boolean, E2, R2>,
    options?: {
      readonly excludeLast?: boolean | undefined
    }
  ): Stream<A, E | E2, R | R2>
} = dual((args) => isStream(args[0]), <A, E, R, E2, R2>(
  self: Stream<A, E, R>,
  predicate: (a: A, n: number) => Effect.Effect<boolean, E2, R2>,
  options?: {
    readonly excludeLast?: boolean | undefined
  }
): Stream<A, E | E2, R | R2> =>
  transformPull(self, (pull, _scope) =>
    Effect.sync(() => {
      let i = 0
      let done = false
      return Effect.gen(function*() {
        if (done) return yield* Cause.done()
        const chunk = yield* pull
        for (let j = 0; j < chunk.length; j++) {
          if (yield* predicate(chunk[j], i++)) {
            done = true
            const arr = chunk.slice(0, options?.excludeLast ? j : j + 1)
            return Arr.isReadonlyArrayNonEmpty(arr) ? arr : yield* Cause.done()
          }
        }
        return chunk
      })
    })))
Referenced by 1 symbols