Class: Sbmt::Outbox::V2::PollThrottler::RedisQueueSize

Inherits:
Base show all
Defined in:
lib/sbmt/outbox/v2/poll_throttler/redis_queue_size.rb

Instance Method Summary collapse

Methods inherited from Base

#call

Methods inherited from DryInteractor

call

Constructor Details

#initialize(redis:, min_size: -1,, max_size: 100, delay: 0) ⇒ RedisQueueSize

Returns a new instance of RedisQueueSize.



12
13
14
15
16
17
18
19
 
# File 'lib/sbmt/outbox/v2/poll_throttler/redis_queue_size.rb', line 12

def initialize(redis:, min_size: -1, max_size: 100, delay: 0)
  super()

  @redis = redis
  @min_size = min_size
  @max_size = max_size
  @delay = delay
end

Instance Method Details

#wait(worker_num, poll_task, _task_result) ⇒ Object 



21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
 
# File 'lib/sbmt/outbox/v2/poll_throttler/redis_queue_size.rb', line 21

def wait(worker_num, poll_task, _task_result)
  # LLEN is O(1)
  queue_size = @redis.call("LLEN", poll_task.redis_queue).to_i
  redis_job_queue_size.set(metric_tags(poll_task), queue_size)

  if queue_size < @min_size
    # just throttle (not skip) to wait for job queue size becomes acceptable
    sleep(@delay)
    return Success(Sbmt::Outbox::V2::Throttler::THROTTLE_STATUS)
  end

  if queue_size > @max_size
    # completely skip poll-cycle if job queue is oversized
    sleep(@delay)
    return Success(Sbmt::Outbox::V2::Throttler::SKIP_STATUS)
  end

  Success(Sbmt::Outbox::V2::Throttler::NOOP_STATUS)
rescue => ex
  Failure(ex.message)
end