Module: Resque::Scheduler::DelayingExtensions
- Included in:
- Extension
- Defined in:
- lib/resque/scheduler/delaying_extensions.rb
Instance Method Summary collapse
- #count_all_scheduled_jobs ⇒ Object
-
#delayed?(klass, *args) ⇒ Boolean
Discover if a job has been delayed.
-
#delayed_push(timestamp, item) ⇒ Object
Used internally to stuff the item into the schedule sorted list.
-
#delayed_queue_peek(start, count) ⇒ Object
Returns an array of timestamps based on start and count.
-
#delayed_queue_schedule_size ⇒ Object
Returns the size of the delayed queue schedule this does not represent the number of items in the queue to be scheduled.
-
#delayed_timestamp_peek(timestamp, start, count) ⇒ Object
Returns an array of delayed items for the given timestamp.
-
#delayed_timestamp_size(timestamp) ⇒ Object
Returns the number of jobs for a given timestamp in the delayed queue schedule.
-
#enqueue_at(timestamp, klass, *args) ⇒ Object
This method is nearly identical to
enqueue
only it also takes a timestamp which will be used to schedule the job for queueing. -
#enqueue_at_with_queue(queue, timestamp, klass, *args) ⇒ Object
Identical to
enqueue_at
, except you can also specify a queue in which the job will be placed after the timestamp has passed. -
#enqueue_delayed(klass, *args) ⇒ Object
Given an encoded item, enqueue it now.
-
#enqueue_delayed_selection(klass = nil) ⇒ Object
Given a block, enqueue jobs now that return true from a block.
- #enqueue_delayed_with_queue(klass, queue, *args) ⇒ Object
-
#enqueue_in(number_of_seconds_from_now, klass, *args) ⇒ Object
Identical to enqueue_at but takes number_of_seconds_from_now instead of a timestamp.
-
#enqueue_in_with_queue(queue, number_of_seconds_from_now, klass, *args) ⇒ Object
Identical to
enqueue_in
, except you can also specify a queue in which the job will be placed after the number of seconds has passed. -
#find_delayed_selection(klass = nil, &block) ⇒ Object
Given a block, find jobs that return true from a block.
- #get_last_enqueued_at(job_name) ⇒ Object
- #last_enqueued_at(job_name, date) ⇒ Object
-
#next_delayed_timestamp(at_time = nil) ⇒ Object
Returns the next delayed queue timestamp (don’t call directly).
-
#next_item_for_timestamp(timestamp) ⇒ Object
Returns the next item to be processed for a given timestamp, nil if done.
-
#remove_delayed(klass, *args) ⇒ Object
Given an encoded item, remove it from the delayed_queue.
- #remove_delayed_in_queue(klass, queue, *args) ⇒ Object
-
#remove_delayed_job_from_timestamp(timestamp, klass, *args) ⇒ Object
Given a timestamp and job (klass + args) it removes all instances and returns the count of jobs removed.
-
#remove_delayed_selection(klass = nil) ⇒ Object
Given a block, remove jobs that return true from a block.
-
#reset_delayed_queue ⇒ Object
Clears all jobs created with enqueue_at or enqueue_in.
-
#scheduled_at(klass, *args) ⇒ Object
Returns delayed jobs schedule timestamp for
klass
,args
.
Instance Method Details
#count_all_scheduled_jobs ⇒ Object
252 253 254 255 256 257 258 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 252 def count_all_scheduled_jobs total_jobs = 0 Array(redis.zrange(:delayed_queue_schedule, 0, -1)).each do |ts| total_jobs += redis.llen("delayed:#{ts}").to_i end total_jobs end |
#delayed?(klass, *args) ⇒ Boolean
265 266 267 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 265 def delayed?(klass, *args) !scheduled_at(klass, *args).empty? end |
#delayed_push(timestamp, item) ⇒ Object
Used internally to stuff the item into the schedule sorted list. timestamp
can be either in seconds or a datetime object. The insertion time complexity is O(log(n)). Returns true if it’s the first job to be scheduled at that time, else false.
70 71 72 73 74 75 76 77 78 79 80 81 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 70 def delayed_push(, item) # First add this item to the list for this timestamp redis.rpush("delayed:#{.to_i}", encode(item)) # Store the timestamps at with this item occurs redis.sadd("timestamps:#{encode(item)}", "delayed:#{.to_i}") # Now, add this timestamp to the zsets. The score and the value are # the same since we'll be querying by timestamp, and we don't have # anything else to store. redis.zadd :delayed_queue_schedule, .to_i, .to_i end |
#delayed_queue_peek(start, count) ⇒ Object
Returns an array of timestamps based on start and count
84 85 86 87 88 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 84 def delayed_queue_peek(start, count) result = redis.zrange(:delayed_queue_schedule, start, start + count - 1) Array(result).map(&:to_i) end |
#delayed_queue_schedule_size ⇒ Object
Returns the size of the delayed queue schedule this does not represent the number of items in the queue to be scheduled
92 93 94 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 92 def delayed_queue_schedule_size redis.zcard :delayed_queue_schedule end |
#delayed_timestamp_peek(timestamp, start, count) ⇒ Object
Returns an array of delayed items for the given timestamp
103 104 105 106 107 108 109 110 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 103 def (, start, count) if 1 == count r = list_range "delayed:#{.to_i}", start, count r.nil? ? [] : [r] else list_range "delayed:#{.to_i}", start, count end end |
#delayed_timestamp_size(timestamp) ⇒ Object
Returns the number of jobs for a given timestamp in the delayed queue schedule
98 99 100 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 98 def () redis.llen("delayed:#{.to_i}").to_i end |
#enqueue_at(timestamp, klass, *args) ⇒ Object
This method is nearly identical to enqueue
only it also takes a timestamp which will be used to schedule the job for queueing. Until timestamp is in the past, the job will sit in the schedule list.
13 14 15 16 17 18 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 13 def enqueue_at(, klass, *args) validate(klass) enqueue_at_with_queue( queue_from_class(klass), , klass, *args ) end |
#enqueue_at_with_queue(queue, timestamp, klass, *args) ⇒ Object
Identical to enqueue_at
, except you can also specify a queue in which the job will be placed after the timestamp has passed. It respects Resque.inline option, by creating the job right away instead of adding to the queue.
24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 24 def enqueue_at_with_queue(queue, , klass, *args) return false unless plugin.run_before_schedule_hooks(klass, *args) if Resque.inline? || .to_i <= Time.now.to_i # Just create the job and let resque perform it right away with # inline. If the class is a custom job class, call self#scheduled # on it. This allows you to do things like # Resque.enqueue_at(timestamp, CustomJobClass, :opt1 => val1). # Otherwise, pass off to Resque. if klass.respond_to?(:scheduled) klass.scheduled(queue, klass.to_s, *args) else Resque.enqueue_to(queue, klass, *args) end else delayed_push(, job_to_hash_with_queue(queue, klass, args)) end plugin.run_after_schedule_hooks(klass, *args) end |
#enqueue_delayed(klass, *args) ⇒ Object
Given an encoded item, enqueue it now
159 160 161 162 163 164 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 159 def enqueue_delayed(klass, *args) hash = job_to_hash(klass, args) remove_delayed(klass, *args).times do Resque::Scheduler.enqueue_from_config(hash) end end |
#enqueue_delayed_selection(klass = nil) ⇒ Object
Given a block, enqueue jobs now that return true from a block
This allows for enqueuing of delayed jobs that have arguments matching certain criteria
190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 190 def enqueue_delayed_selection(klass = nil) raise ArgumentError, 'Please supply a block' unless block_given? found_jobs = find_delayed_selection(klass) { |args| yield(args) } found_jobs.reduce(0) do |sum, encoded_job| decoded_job = decode(encoded_job) klass = Util.constantize(decoded_job['class']) queue = decoded_job['queue'] if queue jobs_queued = enqueue_delayed_with_queue(klass, queue, *decoded_job['args']) else jobs_queued = enqueue_delayed(klass, *decoded_job['args']) end jobs_queued + sum end end |
#enqueue_delayed_with_queue(klass, queue, *args) ⇒ Object
166 167 168 169 170 171 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 166 def enqueue_delayed_with_queue(klass, queue, *args) hash = job_to_hash_with_queue(queue, klass, args) remove_delayed_in_queue(klass, queue, *args).times do Resque::Scheduler.enqueue_from_config(hash) end end |
#enqueue_in(number_of_seconds_from_now, klass, *args) ⇒ Object
Identical to enqueue_at but takes number_of_seconds_from_now instead of a timestamp.
47 48 49 50 51 52 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 47 def enqueue_in(number_of_seconds_from_now, klass, *args) unless number_of_seconds_from_now.is_a?(Numeric) raise ArgumentError, 'Please supply a numeric number of seconds' end enqueue_at(Time.now + number_of_seconds_from_now, klass, *args) end |
#enqueue_in_with_queue(queue, number_of_seconds_from_now, klass, *args) ⇒ Object
Identical to enqueue_in
, except you can also specify a queue in which the job will be placed after the number of seconds has passed.
57 58 59 60 61 62 63 64 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 57 def enqueue_in_with_queue(queue, number_of_seconds_from_now, klass, *args) unless number_of_seconds_from_now.is_a?(Numeric) raise ArgumentError, 'Please supply a numeric number of seconds' end enqueue_at_with_queue(queue, Time.now + number_of_seconds_from_now, klass, *args) end |
#find_delayed_selection(klass = nil, &block) ⇒ Object
Given a block, find jobs that return true from a block
This allows for finding of delayed jobs that have arguments matching certain criteria
213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 213 def find_delayed_selection(klass = nil, &block) raise ArgumentError, 'Please supply a block' unless block_given? = redis.zrange(:delayed_queue_schedule, 0, -1) # Beyond 100 there's almost no improvement in speed found = .each_slice(100).map do |ts_group| jobs = redis.pipelined do |r| ts_group.each do |ts| r.lrange("delayed:#{ts}", 0, -1) end end jobs.flatten.select do |payload| payload_matches_selection?(decode(payload), klass, &block) end end found.flatten end |
#get_last_enqueued_at(job_name) ⇒ Object
281 282 283 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 281 def get_last_enqueued_at(job_name) redis.hget('delayed:last_enqueued_at', job_name) end |
#last_enqueued_at(job_name, date) ⇒ Object
277 278 279 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 277 def last_enqueued_at(job_name, date) redis.hset('delayed:last_enqueued_at', job_name, date) end |
#next_delayed_timestamp(at_time = nil) ⇒ Object
Returns the next delayed queue timestamp (don’t call directly)
114 115 116 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 114 def (at_time = nil) (nil, at_time || Time.now) end |
#next_item_for_timestamp(timestamp) ⇒ Object
Returns the next item to be processed for a given timestamp, nil if done. (don’t call directly) timestamp
can either be in seconds or a datetime
121 122 123 124 125 126 127 128 129 130 131 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 121 def () key = "delayed:#{.to_i}" encoded_item = redis.lpop(key) redis.srem("timestamps:#{encoded_item}", key) item = decode(encoded_item) # If the list is empty, remove it. (key, ) item end |
#remove_delayed(klass, *args) ⇒ Object
Given an encoded item, remove it from the delayed_queue
148 149 150 151 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 148 def remove_delayed(klass, *args) search = encode(job_to_hash(klass, args)) remove_delayed_job(search) end |
#remove_delayed_in_queue(klass, queue, *args) ⇒ Object
153 154 155 156 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 153 def remove_delayed_in_queue(klass, queue, *args) search = encode(job_to_hash_with_queue(queue, klass, args)) remove_delayed_job(search) end |
#remove_delayed_job_from_timestamp(timestamp, klass, *args) ⇒ Object
Given a timestamp and job (klass + args) it removes all instances and returns the count of jobs removed.
O(N) where N is the number of jobs scheduled to fire at the given timestamp
239 240 241 242 243 244 245 246 247 248 249 250 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 239 def (, klass, *args) return 0 if Resque.inline? key = "delayed:#{.to_i}" encoded_job = encode(job_to_hash(klass, args)) redis.srem("timestamps:#{encoded_job}", key) count = redis.lrem(key, 0, encoded_job) (key, ) count end |
#remove_delayed_selection(klass = nil) ⇒ Object
Given a block, remove jobs that return true from a block
This allows for removal of delayed jobs that have arguments matching certain criteria
177 178 179 180 181 182 183 184 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 177 def remove_delayed_selection(klass = nil) raise ArgumentError, 'Please supply a block' unless block_given? found_jobs = find_delayed_selection(klass) { |args| yield(args) } found_jobs.reduce(0) do |sum, encoded_job| sum + remove_delayed_job(encoded_job) end end |
#reset_delayed_queue ⇒ Object
Clears all jobs created with enqueue_at or enqueue_in
134 135 136 137 138 139 140 141 142 143 144 145 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 134 def reset_delayed_queue Array(redis.zrange(:delayed_queue_schedule, 0, -1)).each do |item| key = "delayed:#{item}" items = redis.lrange(key, 0, -1) redis.pipelined do items.each { |ts_item| redis.del("timestamps:#{ts_item}") } end redis.del key end redis.del :delayed_queue_schedule end |
#scheduled_at(klass, *args) ⇒ Object
Returns delayed jobs schedule timestamp for klass
, args
.
270 271 272 273 274 275 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 270 def scheduled_at(klass, *args) search = encode(job_to_hash(klass, args)) redis.smembers("timestamps:#{search}").map do |key| key.tr('delayed:', '').to_i end end |