Class: Sidekiq::Scheduled::Poller

Inherits:
Object
  • Object
show all
Includes:
Component
Defined in:
lib/sidekiq/scheduled.rb

Overview

The Poller checks Redis every N seconds for jobs in the retry or scheduled set have passed their timestamp and should be enqueued. If so, it just pops the job back onto its original queue so the workers can pick it up like any other job.

Constant Summary collapse

INITIAL_WAIT = 
10

Instance Attribute Summary collapse

Attributes included from Component

#config

Instance Method Summary collapse

Methods included from Component

#default_tag, #fire_event, #handle_exception, #hostname, #identity, #inspect, #logger, #mono_ms, #process_nonce, #real_ms, #redis, #safe_thread, #tid, #watchdog

Constructor Details

#initialize(config) ⇒ Poller 



77
78
79
80
81
82
83
84
85
 
# File 'lib/sidekiq/scheduled.rb', line 77

def initialize(config)
  @config = config
  @enq = (config[:scheduled_enq] || Sidekiq::Scheduled::Enq).new(config)
  @sleeper = ConnectionPool::TimedStack.new
  @done = false
  @thread = nil
  @count_calls = 0
  @rnd = Random.new
end

Instance Attribute Details

#rnd ⇒ Object

Returns the value of attribute rnd.



75
76
77
 
# File 'lib/sidekiq/scheduled.rb', line 75

def rnd
  @rnd
end

Instance Method Details

#enqueue ⇒ Object 



108
109
110
111
112
113
114
115
 
# File 'lib/sidekiq/scheduled.rb', line 108

def enqueue
  @enq.enqueue_jobs
rescue => ex
  # Most likely a problem with redis networking.
  # Punt and try again at the next interval
  logger.error ex.message
  handle_exception(ex)
end

#start ⇒ Object 



96
97
98
99
100
101
102
103
104
105
106
 
# File 'lib/sidekiq/scheduled.rb', line 96

def start
  @thread ||= safe_thread("scheduler") {
    initial_wait

    until @done
      enqueue
      wait
    end
    logger.info("Scheduler exiting...")
  }
end

#terminate ⇒ Object

Shut down this instance, will pause until the thread is dead.



88
89
90
91
92
93
94
 
# File 'lib/sidekiq/scheduled.rb', line 88

def terminate
  @done = true
  @enq.terminate

  @sleeper << 0
  @thread&.value
end