Hyperlinkv0.8.0-beta.28

Stream

Stream.runForEachWhileconsteffect/Stream.ts:10916
<A, E2, R2>(f: (a: A) => Effect.Effect<boolean, E2, R2>): <E, R>(
  self: Stream<A, E, R>
) => Effect.Effect<void, E2 | E, R2 | R>
<A, E, R, E2, R2>(
  self: Stream<A, E, R>,
  f: (a: A) => Effect.Effect<boolean, E2, R2>
): Effect.Effect<void, E | E2, R | R2>

Runs the stream, applying the effectful predicate to each element and stopping when it returns false.

Example (Running effects while a predicate holds)

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

const program = Effect.gen(function*() {
  const stream = Stream.make(1, 2, 3, 4, 5)

  yield* Stream.runForEachWhile(stream, (n) =>
    Effect.gen(function*() {
      yield* Console.log(`Processing: ${n}`)
      return n < 3
    })
  )
})

Effect.runPromise(program)
// Processing: 1
// Processing: 2
// Processing: 3
destructors
Source effect/Stream.ts:1091627 lines
export const runForEachWhile: {
  <A, E2, R2>(
    f: (a: A) => Effect.Effect<boolean, E2, R2>
  ): <E, R>(self: Stream<A, E, R>) => Effect.Effect<void, E2 | E, R2 | R>
  <A, E, R, E2, R2>(
    self: Stream<A, E, R>,
    f: (a: A) => Effect.Effect<boolean, E2, R2>
  ): Effect.Effect<void, E | E2, R | R2>
} = dual(2, <A, E, R, E2, R2>(
  self: Stream<A, E, R>,
  f: (a: A) => Effect.Effect<boolean, E2, R2>
): Effect.Effect<void, E | E2, R | R2> =>
  Channel.runForEachWhile(self.channel, (arr) => {
    let done = false
    let i = 0
    return Effect.map(
      Effect.whileLoop({
        while: () => !done && i < arr.length,
        body: () => f(arr[i]),
        step(b) {
          i++
          if (!b) done = true
        }
      }),
      () => !done
    )
  }))