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.
-
#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_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_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
231 232 233 234 235 236 237 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 231 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
244 245 246 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 244 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 Insertion if 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
91 92 93 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 91 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
102 103 104 105 106 107 108 109 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 102 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
97 98 99 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 97 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::Job.create(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
153 154 155 156 157 158 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 153 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
177 178 179 180 181 182 183 184 185 186 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 177 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']) sum + enqueue_delayed(klass, *decoded_job['args']) 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
192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 192 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
260 261 262 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 260 def get_last_enqueued_at(job_name) redis.hget('delayed:last_enqueued_at', job_name) end |
#last_enqueued_at(job_name, date) ⇒ Object
256 257 258 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 256 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)
113 114 115 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 113 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
120 121 122 123 124 125 126 127 128 129 130 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 120 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
147 148 149 150 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 147 def remove_delayed(klass, *args) search = encode(job_to_hash(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
218 219 220 221 222 223 224 225 226 227 228 229 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 218 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
164 165 166 167 168 169 170 171 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 164 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
133 134 135 136 137 138 139 140 141 142 143 144 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 133 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
.
249 250 251 252 253 254 |
# File 'lib/resque/scheduler/delaying_extensions.rb', line 249 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 |