Class: Sbmt::Outbox::V2::PollThrottler::RedisQueueTimeLag

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

Instance Method Summary collapse

Methods inherited from Base

#call

Methods inherited from DryInteractor

call

Constructor Details

#initialize(redis:, min_lag: 5, delay: 0) ⇒ RedisQueueTimeLag

Returns a new instance of RedisQueueTimeLag.



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

def initialize(redis:, min_lag: 5, delay: 0)
  super()

  @redis = redis
  @min_lag = min_lag
  @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
 
# File 'lib/sbmt/outbox/v2/poll_throttler/redis_queue_time_lag.rb', line 21

def wait(worker_num, poll_task, _task_result)
  # LINDEX is O(1) for first/last element
  oldest_job = @redis.call("LINDEX", poll_task.redis_queue, -1)
  return Success(Sbmt::Outbox::V2::Throttler::NOOP_STATUS) if oldest_job.nil?

  job = RedisJob.deserialize!(oldest_job)
  time_lag = Time.current.to_i - job.timestamp

  redis_job_queue_time_lag.set(metric_tags(poll_task), time_lag)

  if time_lag <= @min_lag
    sleep(@delay)
    return Success(Sbmt::Outbox::V2::Throttler::SKIP_STATUS)
  end

  Success(Sbmt::Outbox::V2::Throttler::NOOP_STATUS)
rescue => ex
  # noop, just skip any redis / serialization errors
  Failure(ex.message)
end