Stream.paginate — API

Stream.paginateconsteffect/Stream.ts:1684

<S, A, E = never, R = never>(
  s: S,
  f: (
    s: S
  ) => Effect.Effect<readonly [ReadonlyArray<A>, Option.Option<S>], E, R>
): Stream<A, E, R>

Creates a stream by repeatedly evaluating an effectful page function.

When to use

Use to consume paginated APIs where each step returns a batch of values together with an optional next state.

Details

This is similar to unfold, but each step can emit zero or more values and independently decide whether another state should be requested.

Example (Paginating stream state)

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

const stream = Stream.paginate(0, (n: number) =>
  Effect.succeed(
    [
      [n],
      n < 3 ? Option.some(n + 1) : Option.none<number>()
    ] as const
  ))

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
// Output: [ 0, 1, 2, 3 ]

constructorsunfold

Source effect/Stream.ts:168422 lines

export const paginate = <S, A, E = never, R = never>(
  s: S,
  f: (
    s: S
  ) => Effect.Effect<readonly [ReadonlyArray<A>, Option.Option<S>], E, R>
): Stream<A, E, R> =>
  fromPull(Effect.sync(() => {
    let state = s
    let done = false
    return Effect.suspend(function loop(): Pull.Pull<Arr.NonEmptyReadonlyArray<A>, E, void, R> {
      if (done) return Cause.done()
      return Effect.flatMap(f(state), ([a, s]) => {
        if (Option.isNone(s)) {
          done = true
        } else {
          state = s.value
        }
        if (!Arr.isReadonlyArrayNonEmpty(a)) return loop()
        return Effect.succeed(a)
      })
    })
  }))