Module: Resque::StuckQueue
- Defined in:
- lib/resque_stuck_queue.rb,
lib/resque_stuck_queue/config.rb,
lib/resque_stuck_queue/signals.rb,
lib/resque_stuck_queue/version.rb,
lib/resque_stuck_queue/heartbeat_job.rb
Defined Under Namespace
Modules: Signals
Classes: Config, HeartbeatJob
Constant Summary
collapse
- HEARTBEAT_KEY =
"resque-stuck-queue"
- TRIGGERED_KEY =
"resque-stuck-queue-last-triggered"
- HEARTBEAT_TIMEOUT =
check/refresh every 20 mins.
20 * 60
- TRIGGER_TIMEOUT =
warn/trigger after an hour (with 20 min heartbeat time).
40 * 60
- LOGGER =
Logger.new($stdout)
- TRIGGERED_HANDLER =
must be called by convention: type_handler
proc { |queue_name, lag| Resque::StuckQueue::LOGGER.info("Shit gone bad with them queues...on #{queue_name}. Lag time is #{lag}") }
- RECOVERED_HANDLER =
proc { |queue_name, lag| Resque::StuckQueue::LOGGER.info("recovered queue phew #{queue_name}. Lag time is #{lag}") }
- VERSION =
"0.3.2"
Class Attribute Summary collapse
Class Method Summary
collapse
Class Attribute Details
.config ⇒ Object
Returns the value of attribute config.
17
18
19
|
# File 'lib/resque_stuck_queue.rb', line 17
def config
@config
end
|
Class Method Details
.force_stop! ⇒ Object
98
99
100
101
102
|
# File 'lib/resque_stuck_queue.rb', line 98
def force_stop!
logger.info("Force stopping")
@threads.map(&:kill)
reset!
end
|
.heartbeat_key_for(queue) ⇒ Object
32
33
34
35
36
37
38
|
# File 'lib/resque_stuck_queue.rb', line 32
def heartbeat_key_for(queue)
if config[:heartbeat_key]
"#{queue}:#{config[:heartbeat_key]}"
else
"#{queue}:#{HEARTBEAT_KEY}"
end
end
|
.heartbeat_keys ⇒ Object
48
49
50
|
# File 'lib/resque_stuck_queue.rb', line 48
def heartbeat_keys
queues.map{|q| heartbeat_key_for(q) }
end
|
.logger ⇒ Object
23
24
25
|
# File 'lib/resque_stuck_queue.rb', line 23
def logger
@logger ||= (config[:logger] || StuckQueue::LOGGER)
end
|
.queues ⇒ Object
52
53
54
|
# File 'lib/resque_stuck_queue.rb', line 52
def queues
@queues ||= (config[:queues] || [:app])
end
|
.redis ⇒ Object
27
28
29
30
|
# File 'lib/resque_stuck_queue.rb', line 27
def redis
@redis ||= config[:redis]
HeartbeatJob.redis = @redis
end
|
.reset! ⇒ Object
104
105
106
107
108
109
110
|
# File 'lib/resque_stuck_queue.rb', line 104
def reset!
@config = Config.new @queues = nil
@running = false
@logger = nil
end
|
.reset_keys ⇒ Object
112
113
114
115
116
117
|
# File 'lib/resque_stuck_queue.rb', line 112
def reset_keys
queues.each do |qn|
redis.del(heartbeat_key_for(qn))
redis.del(triggered_key_for(qn))
end
end
|
.start ⇒ Object
call this after setting config. once started you should’t be allowed to modify it
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
|
# File 'lib/resque_stuck_queue.rb', line 64
def start
@running = true
@stopped = false
@threads = []
config.validate_required_keys!
config.freeze
Signals.enable!
log_starting_info
reset_keys
Redis::Classy.db = redis if Redis::Classy.db.nil?
setup_heartbeat_thread
setup_watcher_thread
@threads.map(&:join)
logger.info("threads stopped")
@stopped = true
end
|
.start_in_background ⇒ Object
56
57
58
59
60
61
|
# File 'lib/resque_stuck_queue.rb', line 56
def start_in_background
Thread.new do
Thread.current.abort_on_exception = config[:abort_on_exception]
self.start
end
end
|
.stop ⇒ Object
89
90
91
92
93
94
95
96
|
# File 'lib/resque_stuck_queue.rb', line 89
def stop
reset!
while @stopped == false
sleep 1
end
logger.info("Stopped")
end
|
.stopped? ⇒ Boolean
119
120
121
|
# File 'lib/resque_stuck_queue.rb', line 119
def stopped?
@stopped
end
|
.triggered_key_for(queue) ⇒ Object
40
41
42
43
44
45
46
|
# File 'lib/resque_stuck_queue.rb', line 40
def triggered_key_for(queue)
if config[:triggered_key]
"#{queue}:#{self.config[:triggered_key]}"
else
"#{queue}:#{TRIGGERED_KEY}"
end
end
|