Hyperlinkv0.8.0-beta.28

← Channel

Channel.mapAccumconsteffect/Channel.ts:3590
<S, OutElem, B, E = never, R = never>(
  initial: LazyArg<S>,
  f: (
    s: S,
    a: Types.NoInfer<OutElem>
  ) =>
    | Effect.Effect<readonly [state: S, values: ReadonlyArray<B>], E, R>
    | readonly [state: S, values: ReadonlyArray<B>],
  options?: { readonly onHalt?: ((state: S) => Array<B>) | undefined }
): <OutErr, OutDone, InElem, InErr, InDone, Env>(
  self: Channel<OutElem, OutErr, OutDone, InElem, InErr, InDone, Env>
) => Channel<B, OutErr | E, OutDone, InElem, InErr, InDone, Env | R>
<
  OutElem,
  OutErr,
  OutDone,
  InElem,
  InErr,
  InDone,
  Env,
  S,
  B,
  E = never,
  R = never
>(
  self: Channel<OutElem, OutErr, OutDone, InElem, InErr, InDone, Env>,
  initial: LazyArg<S>,
  f: (
    s: S,
    a: Types.NoInfer<OutElem>
  ) =>
    | Effect.Effect<readonly [state: S, values: ReadonlyArray<B>], E, R>
    | readonly [state: S, values: ReadonlyArray<B>],
  options?: { readonly onHalt?: ((state: S) => Array<B>) | undefined }
): Channel<B, OutErr | E, OutDone, InElem, InErr, InDone, Env | R>

Maps over a channel statefully with an accumulator, where each element can produce multiple output values.

Example (Mapping with accumulated state)

import { Channel, Effect } from "effect"

// Create a channel with numbers
const numbersChannel = Channel.fromIterable([1, 2, 3, 4])

// Use mapAccum to create running sums and emit both current and sum
const runningSum = Channel.mapAccum(
  numbersChannel,
  () => 0, // initial accumulator state
  (sum, current) => {
    const newSum = sum + current
    // Return [newState, outputValues]
    return [newSum, [current, newSum]] as const
  }
)
// Outputs: 1, 1, 2, 3, 3, 6, 4, 10

// Using with Effect for async processing
const asyncMapAccum = Channel.mapAccum(
  numbersChannel,
  () => "",
  (acc, value) =>
    Effect.gen(function*() {
      const newAcc = acc + value.toString()
      return [newAcc, [`${value}-processed`, newAcc]] as const
    })
)
sequencing
Source effect/Channel.ts:3590102 lines
export const mapAccum: {
  <S, OutElem, B, E = never, R = never>(
    initial: LazyArg<S>,
    f: (
      s: S,
      a: Types.NoInfer<OutElem>
    ) =>
      | Effect.Effect<readonly [state: S, values: ReadonlyArray<B>], E, R>
      | readonly [state: S, values: ReadonlyArray<B>],
    options?: {
      readonly onHalt?: ((state: S) => Array<B>) | undefined
    }
  ): <
    OutErr,
    OutDone,
    InElem,
    InErr,
    InDone,
    Env
  >(self: Channel<OutElem, OutErr, OutDone, InElem, InErr, InDone, Env>) => Channel<
    B,
    OutErr | E,
    OutDone,
    InElem,
    InErr,
    InDone,
    Env | R
  >
  <OutElem, OutErr, OutDone, InElem, InErr, InDone, Env, S, B, E = never, R = never>(
    self: Channel<OutElem, OutErr, OutDone, InElem, InErr, InDone, Env>,
    initial: LazyArg<S>,
    f: (
      s: S,
      a: Types.NoInfer<OutElem>
    ) =>
      | Effect.Effect<readonly [state: S, values: ReadonlyArray<B>], E, R>
      | readonly [state: S, values: ReadonlyArray<B>],
    options?: {
      readonly onHalt?: ((state: S) => Array<B>) | undefined
    }
  ): Channel<B, OutErr | E, OutDone, InElem, InErr, InDone, Env | R>
} = dual(
  (args) => isChannel(args[0]),
  <OutElem, OutErr, OutDone, InElem, InErr, InDone, Env, S, B, E = never, R = never>(
    self: Channel<OutElem, OutErr, OutDone, InElem, InErr, InDone, Env>,
    initial: LazyArg<S>,
    f: (
      s: S,
      a: Types.NoInfer<OutElem>
    ) =>
      | Effect.Effect<readonly [state: S, values: ReadonlyArray<B>], E, R>
      | readonly [state: S, values: ReadonlyArray<B>],
    options?: {
      readonly onHalt?: ((state: S) => ReadonlyArray<B>) | undefined
    }
  ): Channel<B, OutErr | E, OutDone, InElem, InErr, InDone, Env | R> =>
    fromTransform((upstream, scope) =>
      Effect.map(toTransform(self)(upstream, scope), (pull) => {
        let state = initial()
        let current: ReadonlyArray<B> | undefined
        let index = 0
        let cause: Cause.Cause<OutErr | Cause.Done<OutDone>> | undefined
        const pullNext = Effect.matchCauseEffect(pull, {
          onFailure(cause_) {
            cause = cause_
            const b = options?.onHalt && options.onHalt(state)
            return b && b.length > 0
              ? Effect.succeed([state, b] as const)
              : Effect.failCause(cause_)
          },
          onSuccess(a): Effect.Effect<readonly [state: S, values: ReadonlyArray<B>], E, R> {
            const b = f(state, a)
            return Arr.isArray(b)
              ? Effect.succeed(b as any)
              : b as any
          }
        })
        const pump = Effect.suspend(function loop(): Pull.Pull<B, OutErr | E, OutDone, R> {
          if (current === undefined) {
            if (cause) return Effect.failCause(cause)
            return Effect.flatMap(pullNext, ([newState, values]) => {
              state = newState
              if (values.length === 0) {
                return loop()
              } else if (values.length === 1) {
                return Effect.succeed(values[0])
              }
              current = values
              return loop()
            })
          }
          const next = current[index++]
          if (index >= current.length) {
            current = undefined
            index = 0
          }
          return Effect.succeed(next)
        })
        return pump
      })
    )
)
Referenced by 5 symbols
Stream.mapAccumStream.mapAccumArrayStream.mapAccumArrayEffectStream.mapAccumEffectStream.scan