Hyperlinkv0.8.0-beta.28

Stream

Stream.debounceconsteffect/Stream.ts:7839
(duration: Duration.Input): <A, E, R>(
  self: Stream<A, E, R>
) => Stream<A, E, R>
<A, E, R>(self: Stream<A, E, R>, duration: Duration.Input): Stream<
  A,
  E,
  R
>

Drops earlier elements within the debounce window and emits only the latest element after the pause.

Example (Debouncing stream elements)

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

const stream = Stream.make(1, 2, 3).pipe(
  Stream.concat(Stream.fromEffect(Effect.sleep(Duration.millis(50)).pipe(Effect.as(4)))),
  Stream.concat(Stream.make(5)),
  Stream.debounce(Duration.millis(30))
)

const program = Effect.gen(function*() {
  const values = yield* Stream.runCollect(stream)
  yield* Console.log(values)
  // Output: [ 3, 5 ]
})
rate limiting
Source effect/Stream.ts:783972 lines
export const debounce: {
  (duration: Duration.Input): <A, E, R>(self: Stream<A, E, R>) => Stream<A, E, R>
  <A, E, R>(self: Stream<A, E, R>, duration: Duration.Input): Stream<A, E, R>
} = dual(
  2,
  <A, E, R>(self: Stream<A, E, R>, duration: Duration.Input): Stream<A, E, R> =>
    transformPull(
      self,
      Effect.fnUntraced(function*(pull, scope) {
        const clock = yield* Clock
        const durationMs = Duration.toMillis(Duration.fromInputUnsafe(duration))
        let lastArr: Arr.NonEmptyReadonlyArray<A> | undefined
        let cause: Cause.Cause<Cause.Done | E> | undefined
        let emitAtMs = Infinity
        const pullLatch = Latch.makeUnsafe()
        const emitLatch = Latch.makeUnsafe()
        const endLatch = Latch.makeUnsafe()

        yield* pull.pipe(
          pullLatch.whenOpen,
          Effect.flatMap((arr) => {
            emitLatch.openUnsafe()
            lastArr = arr
            emitAtMs = clock.currentTimeMillisUnsafe() + durationMs
            return Effect.void
          }),
          Effect.forever({ disableYield: true }),
          Effect.onError((cause_) => {
            cause = cause_
            emitAtMs = clock.currentTimeMillisUnsafe()
            emitLatch.openUnsafe()
            endLatch.openUnsafe()
            return Effect.void
          }),
          Effect.forkIn(scope)
        )

        const sleepLoop = Effect.suspend(function loop(): Pull.Pull<Arr.NonEmptyReadonlyArray<A>, E, void, R> {
          const now = clock.currentTimeMillisUnsafe()
          const timeMs = emitAtMs < now ? durationMs : Math.min(durationMs, emitAtMs - now)
          return Effect.flatMap(Effect.raceFirst(Effect.sleep(timeMs), endLatch.await), () => {
            const now = clock.currentTimeMillisUnsafe()
            if (now < emitAtMs) {
              return loop()
            } else if (lastArr) {
              emitLatch.closeUnsafe()
              pullLatch.closeUnsafe()
              const eff = Effect.succeed(Arr.of(Arr.lastNonEmpty(lastArr)))
              lastArr = undefined
              return eff
            } else if (cause) {
              return Effect.failCause(cause!)
            }
            return loop()
          })
        })

        return Effect.suspend(() => {
          if (cause) {
            if (lastArr) {
              const eff = Effect.succeed(Arr.of(Arr.lastNonEmpty(lastArr)))
              lastArr = undefined
              return eff
            }
            return Effect.failCause(cause)
          }
          pullLatch.openUnsafe()
          return emitLatch.whenOpen(sleepLoop)
        })
      })
    )
)