Hyperlinkv0.8.0-beta.28

← Stream

Stream.mapAccumEffectconsteffect/Stream.ts:7583
<S, A, B, E2, R2>(
  initial: LazyArg<S>,
  f: (
    s: S,
    a: A
  ) => Effect.Effect<
    readonly [state: S, values: ReadonlyArray<B>],
    E2,
    R2
  >,
  options?: {
    readonly onHalt?: ((state: S) => ReadonlyArray<B>) | undefined
  }
): <E, R>(self: Stream<A, E, R>) => Stream<B, E | E2, R | R2>
<A, E, R, S, B, E2, R2>(
  self: Stream<A, E, R>,
  initial: LazyArg<S>,
  f: (
    s: S,
    a: A
  ) => Effect.Effect<
    readonly [state: S, values: ReadonlyArray<B>],
    E2,
    R2
  >,
  options?: {
    readonly onHalt?: ((state: S) => ReadonlyArray<B>) | undefined
  }
): Stream<B, E | E2, R | R2>

Maps each element statefully and effectfully, emitting zero or more output values per input.

When to use

Use when stateful element mapping needs Effects or can fail while emitting zero or more values per input element.

Details

The mapping effect receives the current state and element, then returns the next state plus the values to emit. The state is threaded through the stream.

Example (Effectfully mapping stream values with state)

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

const program = Effect.gen(function*() {
  const result = yield* Stream.make(1, 1, 1).pipe(
    Stream.mapAccumEffect(() => 0, (total, n) =>
      Effect.succeed([total + n, [total + n]])
    ),
    Stream.runCollect
  )

  yield* Console.log(result)
})

Effect.runPromise(program)
// Output: [ 1, 2, 3 ]
mapping
Source effect/Stream.ts:758347 lines
export const mapAccumEffect: {
  <S, A, B, E2, R2>(
    initial: LazyArg<S>,
    f: (s: S, a: A) => Effect.Effect<readonly [state: S, values: ReadonlyArray<B>], E2, R2>,
    options?: {
      readonly onHalt?: ((state: S) => ReadonlyArray<B>) | undefined
    }
  ): <E, R>(self: Stream<A, E, R>) => Stream<B, E | E2, R | R2>
  <A, E, R, S, B, E2, R2>(
    self: Stream<A, E, R>,
    initial: LazyArg<S>,
    f: (s: S, a: A) => Effect.Effect<readonly [state: S, values: ReadonlyArray<B>], E2, R2>,
    options?: {
      readonly onHalt?: ((state: S) => ReadonlyArray<B>) | undefined
    }
  ): Stream<B, E | E2, R | R2>
} = dual((args) => isStream(args[0]), <A, E, R, S, B, E2, R2>(
  self: Stream<A, E, R>,
  initial: LazyArg<S>,
  f: (s: S, a: A) => Effect.Effect<readonly [state: S, values: ReadonlyArray<B>], E2, R2>,
  options?: {
    readonly onHalt?: ((state: S) => ReadonlyArray<B>) | undefined
  }
): Stream<B, E | E2, R | R2> =>
  self.channel.pipe(
    Channel.flattenArray,
    Channel.mapAccum(
      initial,
      (state, a) =>
        Effect.map(
          f(state, a),
          ([state, values]) => [
            state,
            Arr.isReadonlyArrayNonEmpty(values) ? Arr.of(values) : Arr.empty<Arr.NonEmptyReadonlyArray<B>>()
          ]
        ),
      options?.onHalt ?
        {
          onHalt(state) {
            const arr = options.onHalt!(state)
            return Arr.isReadonlyArrayNonEmpty(arr) ? Arr.of(arr) : emptyArr
          }
        } :
        undefined
    ),
    fromChannel
  ))