Module: Sidekiq::LimitFetch

Extended by:
LimitFetch
Included in:
LimitFetch
Defined in:
lib/sidekiq/limit_fetch.rb

Defined Under Namespace

Modules: Global, Instances, Queues Classes: UnitOfWork

Constant Summary collapse

TIMEOUT = 
Sidekiq::BasicFetch::TIMEOUT
RedisBaseConnectionError = 
post_7? ? RedisClient::ConnectionError : Redis::BaseConnectionError
RedisCommandError = 
post_7? ? RedisClient::CommandError : Redis::CommandError

Instance Method Summary collapse

Instance Method Details

#bulk_requeue(*args) ⇒ Object

Backwards compatibility for sidekiq v6.1.0

See Also:



45
46
47
48
49
50
51
 
# File 'lib/sidekiq/limit_fetch.rb', line 45

def bulk_requeue(*args)
  if Sidekiq::BasicFetch.respond_to?(:bulk_requeue) # < 6.1.0
    Sidekiq::BasicFetch.bulk_requeue(*args)
  else # 6.1.0+
    Sidekiq::BasicFetch.new(config).bulk_requeue(*args)
  end
end

#config ⇒ Object 



38
39
40
41
 
# File 'lib/sidekiq/limit_fetch.rb', line 38

def config
  # Post 6.5, Sidekiq.options is deprecated and replaced with passing Sidekiq directly
  post_6_5? ? Sidekiq : Sidekiq.options
end

#new(_) ⇒ Object 



28
29
30
 
# File 'lib/sidekiq/limit_fetch.rb', line 28

def new(_)
  self
end

#post_7? ⇒ Boolean

Returns:

  • (Boolean) 


21
22
23
 
# File 'lib/sidekiq/limit_fetch.rb', line 21

def post_7?
  @post_7 ||= Gem::Version.new(Sidekiq::VERSION) >= Gem::Version.new('7.0.0')
end

#redis_retryable ⇒ Object 



53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
 
# File 'lib/sidekiq/limit_fetch.rb', line 53

def redis_retryable
  yield
rescue RedisBaseConnectionError
  sleep TIMEOUT
  retry
rescue RedisCommandError => error
  # If Redis was restarted and is still loading its snapshot,
  # then we should treat this as a temporary connection error too.
  if error.message =~ /^LOADING/
    sleep TIMEOUT
    retry
  else
    raise
  end
end

#retrieve_work ⇒ Object 



32
33
34
35
36
 
# File 'lib/sidekiq/limit_fetch.rb', line 32

def retrieve_work
  queue, job = redis_brpop(Queues.acquire)
  Queues.release_except(queue)
  UnitOfWork.new(queue, job) if job
end