Hyperlinkv0.8.0-beta.28

← Stream

Stream.mapAccumArrayEffectconsteffect/Stream.ts:7671
<S, A, B, E2, R2>(
  initial: LazyArg<S>,
  f: (
    s: S,
    a: Arr.NonEmptyReadonlyArray<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: Arr.NonEmptyReadonlyArray<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 non-empty input chunk statefully and effectfully, emitting zero or more output values per chunk.

When to use

Use when stateful mapping should process each emitted non-empty chunk with an Effect instead of each element separately.

Details

The mapping effect receives the current state and chunk, then returns the next state plus the values to emit. The state is threaded across chunks.

Example (Effectfully mapping stream chunks with state)

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

const program = Effect.gen(function*() {
  const totals = yield* Stream.make(1, 2, 3, 4).pipe(
    Stream.rechunk(2),
    Stream.mapAccumArrayEffect(() => 0, (total, chunk) =>
      Effect.gen(function*() {
        const next = chunk.reduce((sum, value) => sum + value, total)
        return [next, [next]] as const
      })
    ),
    Stream.runCollect
  )
  yield* Console.log(totals)
})

Effect.runPromise(program)
// Output: [ 3, 10 ]
mapping
Source effect/Stream.ts:767146 lines
export const mapAccumArrayEffect: {
  <S, A, B, E2, R2>(
    initial: LazyArg<S>,
    f: (s: S, a: Arr.NonEmptyReadonlyArray<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: Arr.NonEmptyReadonlyArray<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), <A, E, R, S, B, E2, R2>(
  self: Stream<A, E, R>,
  initial: LazyArg<S>,
  f: (s: S, a: Arr.NonEmptyReadonlyArray<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.mapAccum(
      initial,
      (state, a) =>
        Effect.map(
          f(state, a),
          ([state, values]) => [
            state,
            Arr.isReadonlyArrayNonEmpty(values) ? Arr.of(values) : emptyArr
          ]
        ),
      options?.onHalt ?
        {
          onHalt(state) {
            const arr = options.onHalt!(state)
            return Arr.isReadonlyArrayNonEmpty(arr) ? Arr.of(arr) : emptyArr
          }
        } :
        undefined
    ),
    fromChannel
  ))