Class: FiberStream::RateLimiter

Inherits:
Object
  • Object
show all
Defined in:
lib/fiber_stream/rate_limiter.rb

Overview

Scheduler-aware token-bucket rate limiter.

‘rate` permits refill every `per` seconds. `burst` is the maximum token capacity and defaults to `rate`. Immediate permit grants do not require a scheduler. When FiberStream must sleep, the current fiber must be non-blocking with an installed `Fiber.scheduler`.

Defined Under Namespace

Classes: Request

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(rate:, per: 1, burst: nil, &block) ⇒ RateLimiter

Returns a new instance of RateLimiter.



53
54
55
56
57
58
59
60
61
62
63
 
# File 'lib/fiber_stream/rate_limiter.rb', line 53

def initialize(rate:, per: 1, burst: nil, &block)
  self.class.validate_options!(rate:, per:, burst:)

  @rate = rate
  @per = per
  @burst = burst.nil? ? @rate : burst
  @policy = block
  @mutex = Mutex.new
  @tokens = @burst.to_f
  @updated_at = monotonic_now
end

Class Method Details

.validate_options!(rate:, per: 1, burst: nil) ⇒ Object

:nodoc:



14
15
16
17
18
19
 
# File 'lib/fiber_stream/rate_limiter.rb', line 14

def validate_options!(rate:, per: 1, burst: nil) # :nodoc:
  rate = validate_rate(rate)
  validate_duration(:per, per)
  validate_burst(burst.nil? ? rate : burst)
  nil
end

Instance Method Details

#acquire(permits: 1) ⇒ Object

Acquires ‘permits`, waiting when necessary.

Waits are scheduler-backed and non-blocking. Requests larger than ‘burst` are rejected because the local token bucket could never satisfy them.



69
70
71
72
73
74
75
76
77
78
79
 
# File 'lib/fiber_stream/rate_limiter.rb', line 69

def acquire(permits: 1)
  permits = validate_permits(permits)

  if @policy
    acquire_with_policy(permits)
  else
    acquire_with_token_bucket(permits)
  end

  nil
end