Hyperlinkv0.8.0-beta.28

Stream

Stream.groupAdjacentByconsteffect/Stream.ts:8455
<A, K>(f: (a: NoInfer<A>) => K): <E, R>(
  self: Stream<A, E, R>
) => Stream<readonly [K, Arr.NonEmptyArray<A>], E, R>
<A, E, R, K>(self: Stream<A, E, R>, f: (a: NoInfer<A>) => K): Stream<
  readonly [K, Arr.NonEmptyArray<A>],
  E,
  R
>

Groups consecutive elements that have equal keys into non-empty arrays.

When to use

Use when you already have a stream ordered by the grouping key and want to emit each consecutive run as a non-empty array while keeping later non-adjacent runs separate.

Details

The key is computed with f; adjacent elements whose keys are equal by Equal.equals are emitted as one [key, group]. Later non-adjacent runs with the same key are emitted separately.

Source effect/Stream.ts:845553 lines
export const groupAdjacentBy: {
  <A, K>(
    f: (a: NoInfer<A>) => K
  ): <E, R>(self: Stream<A, E, R>) => Stream<readonly [K, Arr.NonEmptyArray<A>], E, R>
  <A, E, R, K>(
    self: Stream<A, E, R>,
    f: (a: NoInfer<A>) => K
  ): Stream<readonly [K, Arr.NonEmptyArray<A>], E, R>
} = dual(2, <A, E, R, K>(
  self: Stream<A, E, R>,
  f: (a: NoInfer<A>) => K
): Stream<readonly [K, Arr.NonEmptyArray<A>], E, R> =>
  transformPull(self, (pull, _scope) =>
    Effect.sync(() => {
      let currentKey: K = undefined as any
      let group: Arr.NonEmptyArray<A> | undefined
      let toEmit = Arr.empty<readonly [K, Arr.NonEmptyArray<A>]>()
      const loop: Pull.Pull<
        Arr.NonEmptyReadonlyArray<readonly [K, Arr.NonEmptyArray<A>]>,
        E
      > = pull.pipe(
        Effect.flatMap((chunk) => {
          for (let i = 0; i < chunk.length; i++) {
            const item = chunk[i]
            const key = f(item)
            if (group === undefined) {
              currentKey = key
              group = [item]
              continue
            } else if (Equal.equals(key, currentKey)) {
              group.push(item)
              continue
            }
            toEmit.push([currentKey, group])
            currentKey = key
            group = [item]
          }
          if (Arr.isArrayNonEmpty(toEmit)) {
            const out = toEmit
            toEmit = []
            return Effect.succeed(out)
          }
          return loop
        })
      )
      let done = false
      return Pull.catchDone(Effect.suspend(() => done ? Cause.done() : loop), () => {
        done = true
        const out = group
        group = undefined
        return out && Arr.isArrayNonEmpty(out) ? Effect.succeed(Arr.of([currentKey, out])) : Cause.done()
      })
    })))