Hyperlinkv0.8.0-beta.28

Stream

Stream.mapAccumArrayconsteffect/Stream.ts:7501
<S, A, B>(
  initial: LazyArg<S>,
  f: (
    s: S,
    a: Arr.NonEmptyReadonlyArray<A>
  ) => readonly [state: S, values: ReadonlyArray<B>],
  options?: {
    readonly onHalt?: ((state: S) => ReadonlyArray<B>) | undefined
  }
): <E, R>(self: Stream<A, E, R>) => Stream<B, E, R>
<A, E, R, S, B>(
  self: Stream<A, E, R>,
  initial: LazyArg<S>,
  f: (
    s: S,
    a: Arr.NonEmptyReadonlyArray<A>
  ) => readonly [state: S, values: ReadonlyArray<B>],
  options?: { readonly onHalt?: ((state: S) => Array<B>) | undefined }
): Stream<B, E, R>

Maps over non-empty chunk arrays statefully, emitting zero or more values per chunk.

Details

The mapping function runs once per chunk and the state is threaded across chunks.

Example (Statefully mapping stream chunks)

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

const program = Effect.gen(function*() {
  const output = yield* Stream.make(1, 2, 3, 4, 5, 6).pipe(
    Stream.rechunk(2),
    Stream.mapAccumArray(() => 0, (sum: number, chunk) => {
      const next = chunk.reduce((acc, n) => acc + n, sum)
      return [next, [next]]
    }),
    Stream.runCollect
  )
  yield* Console.log(output)
})

Effect.runPromise(program)
// Output: [ 3, 10, 21 ]
mapping
Source effect/Stream.ts:750141 lines
export const mapAccumArray: {
  <S, A, B>(
    initial: LazyArg<S>,
    f: (s: S, a: Arr.NonEmptyReadonlyArray<A>) => readonly [state: S, values: ReadonlyArray<B>],
    options?: {
      readonly onHalt?: ((state: S) => ReadonlyArray<B>) | undefined
    }
  ): <E, R>(self: Stream<A, E, R>) => Stream<B, E, R>
  <A, E, R, S, B>(
    self: Stream<A, E, R>,
    initial: LazyArg<S>,
    f: (s: S, a: Arr.NonEmptyReadonlyArray<A>) => readonly [state: S, values: ReadonlyArray<B>],
    options?: {
      readonly onHalt?: ((state: S) => Array<B>) | undefined
    }
  ): Stream<B, E, R>
} = dual((args) => isStream(args[0]), <A, E, R, S, B>(
  self: Stream<A, E, R>,
  initial: LazyArg<S>,
  f: (s: S, a: Arr.NonEmptyReadonlyArray<A>) => readonly [state: S, values: ReadonlyArray<B>],
  options?: {
    readonly onHalt?: ((state: S) => ReadonlyArray<B>) | undefined
  }
): Stream<B, E, R> =>
  fromChannel(Channel.mapAccum(
    self.channel,
    initial,
    (state, arr) => {
      const [newState, values] = f(state, arr)
      state = newState
      return [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
  )))
Referenced by 6 symbols