Hyperlinkv0.8.0-beta.28

← Stream

Stream.scanconsteffect/Stream.ts:7741
<S, A>(initial: S, f: (s: S, a: A) => S): <E, R>(
  self: Stream<A, E, R>
) => Stream<S, E, R>
<A, E, R, S>(
  self: Stream<A, E, R>,
  initial: S,
  f: (s: S, a: A) => S
): Stream<S, E, R>

Accumulates state across the stream, emitting the initial state and each updated state.

Example (Scanning stream state)

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

const program = Effect.gen(function*() {
  const values = yield* Stream.make(1, 2, 3).pipe(
    Stream.scan(0, (acc, n) => acc + n),
    Stream.runCollect
  )
  yield* Console.log(values)
})

Effect.runPromise(program)
// Output: [ 0, 1, 3, 6 ]
Accumulation
Source effect/Stream.ts:774130 lines
export const scan: {
  <S, A>(
    initial: S,
    f: (s: S, a: A) => S
  ): <E, R>(self: Stream<A, E, R>) => Stream<S, E, R>
  <A, E, R, S>(
    self: Stream<A, E, R>,
    initial: S,
    f: (s: S, a: A) => S
  ): Stream<S, E, R>
} = dual(3, <A, E, R, S>(
  self: Stream<A, E, R>,
  initial: S,
  f: (s: S, a: A) => S
): Stream<S, E, R> =>
  suspend(() => {
    let isFirst = true
    return fromChannel(Channel.mapAccum(self.channel, constant(initial), (state, arr) => {
      const states = Arr.empty<S>() as Arr.NonEmptyArray<S>
      if (isFirst) {
        isFirst = false
        states.push(state)
      }
      for (let index = 0; index < arr.length; index++) {
        state = f(state, arr[index])
        states.push(state)
      }
      return [state, Arr.of(states)]
    }))
  }))