Hyperlinkv0.8.0-beta.28

Stream

Stream.changesWithEffectconsteffect/Stream.ts:9313
<A, E2, R2>(f: (x: A, y: A) => Effect.Effect<boolean, E2, R2>): <E, R>(
  self: Stream<A, E, R>
) => Stream<A, E | E2, R | R2>
<A, E, R, E2, R2>(
  self: Stream<A, E, R>,
  f: (x: A, y: A) => Effect.Effect<boolean, E2, R2>
): Stream<A, E | E2, R | R2>

Emits only elements that differ from the previous element, using an effectful equality check.

Details

The predicate runs for each element after the first; returning true treats it as equal and skips it.

Example (Effectfully emitting changed values)

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

const program = Effect.gen(function*() {
  const stream = Stream.make(1, 1, 2, 2, 3, 3).pipe(
    Stream.changesWithEffect((a, b) => Effect.succeed(a === b))
  )
  const result = yield* Stream.runCollect(stream)
  yield* Console.log(result)
})

Effect.runPromise(program)
// { _id: "Chunk", values: [ 1, 2, 3 ] }
Deduplication
Source effect/Stream.ts:931344 lines
export const changesWithEffect: {
  <A, E2, R2>(
    f: (x: A, y: A) => Effect.Effect<boolean, E2, R2>
  ): <E, R>(self: Stream<A, E, R>) => Stream<A, E | E2, R | R2>
  <A, E, R, E2, R2>(
    self: Stream<A, E, R>,
    f: (x: A, y: A) => Effect.Effect<boolean, E2, R2>
  ): Stream<A, E | E2, R | R2>
} = dual(
  2,
  <A, E, R, E2, R2>(
    self: Stream<A, E, R>,
    f: (x: A, y: A) => Effect.Effect<boolean, E2, R2>
  ): Stream<A, E | E2, R | R2> =>
    transformPull(self, (pull, _scope) =>
      Effect.sync(() => {
        let first = true
        let last: A
        return Effect.flatMap(
          pull,
          Effect.fnUntraced(function* loop(arr): Generator<
            Pull.Pull<any, E | E2, void, R2>,
            Arr.NonEmptyReadonlyArray<A>,
            any
          > {
            const out: Array<A> = []
            let i = 0
            if (first) {
              first = false
              last = arr[0]
              i = 1
              out.push(last)
            }
            for (; i < arr.length; i++) {
              const a = arr[i]
              if (yield* f(a, last)) continue
              last = a
              out.push(a)
            }
            return Arr.isArrayNonEmpty(out) ? out : yield* Effect.flatMap(pull, Effect.fnUntraced(loop))
          })
        )
      }))
)