effect-io-ai

Package: effect
Module: Stream

Stream.throttle

Rate-limits stream chunks with a synchronous cost function.

When to use

Use to throttle chunks when each chunk’s cost can be computed synchronously.

Details

Uses a token bucket. The bucket can accumulate up to units + burst tokens, and each chunk consumes the cost returned by cost.

If using the “enforce” strategy, arrays that do not meet the bandwidth constraints are dropped. If using the “shape” strategy, arrays are delayed until they can be emitted without exceeding the bandwidth constraints.

Defaults to the “shape” strategy.

Example (Throttling stream chunks)

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

const stream = Stream.fromSchedule(Schedule.spaced("50 millis")).pipe(
  Stream.take(6),
  Stream.throttle({
    cost: (arr) => arr.length,
    units: 1,
    duration: "100 millis",
    strategy: "shape"
  })
)

const program = Effect.gen(function*() {
  const values = yield* Stream.runCollect(stream)
  yield* Console.log(values)
  // Output: [ 0, 1, 2, 3, 4, 5 ]
})

Signature

declare const throttle: { <A>(options: { readonly cost: (arr: Arr.NonEmptyReadonlyArray<A>) => number; readonly units: number; readonly duration: Duration.Input; readonly burst?: number | undefined; readonly strategy?: "enforce" | "shape" | undefined; }): <E, R>(self: Stream<A, E, R>) => Stream<A, E, R>; <A, E, R>(self: Stream<A, E, R>, options: { readonly cost: (arr: Arr.NonEmptyReadonlyArray<A>) => number; readonly units: number; readonly duration: Duration.Input; readonly burst?: number | undefined; readonly strategy?: "enforce" | "shape" | undefined; }): Stream<A, E, R>; }

Source

Since v2.0.0