class Pallets::Backends::Redis
Constants
- CONTEXT_KEY
- GIVEN_UP_SET_KEY
- JOBMASKS_KEY
- JOBMASK_KEY
- QUEUE_KEY
- RELIABILITY_QUEUE_KEY
- RELIABILITY_SET_KEY
- REMAINING_KEY
- RETRY_SET_KEY
- WORKFLOW_QUEUE_KEY
Public Class Methods
new(blocking_timeout:, failed_job_lifespan:, failed_job_max_count:, job_timeout:, pool_size:, **options)
click to toggle source
# File lib/pallets/backends/redis.rb, line 17 def initialize(blocking_timeout:, failed_job_lifespan:, failed_job_max_count:, job_timeout:, pool_size:, **options) @blocking_timeout = blocking_timeout @failed_job_lifespan = failed_job_lifespan @failed_job_max_count = failed_job_max_count @job_timeout = job_timeout @pool = Pallets::Pool.new(pool_size) { ::Redis.new(options) } register_scripts end
Public Instance Methods
discard(job)
click to toggle source
# File lib/pallets/backends/redis.rb, line 67 def discard(job) @pool.execute do |client| client.evalsha( @scripts['discard'], [GIVEN_UP_SET_KEY, RELIABILITY_QUEUE_KEY, RELIABILITY_SET_KEY], [Time.now.to_f, job, Time.now.to_f - @failed_job_lifespan, @failed_job_max_count] ) end end
get_context(wfid)
click to toggle source
# File lib/pallets/backends/redis.rb, line 41 def get_context(wfid) @pool.execute do |client| client.hgetall(CONTEXT_KEY % wfid) end end
give_up(wfid, job, old_job)
click to toggle source
# File lib/pallets/backends/redis.rb, line 77 def give_up(wfid, job, old_job) @pool.execute do |client| client.evalsha( @scripts['give_up'], [GIVEN_UP_SET_KEY, RELIABILITY_QUEUE_KEY, RELIABILITY_SET_KEY, JOBMASKS_KEY % wfid, WORKFLOW_QUEUE_KEY % wfid, REMAINING_KEY % wfid, CONTEXT_KEY % wfid], [Time.now.to_f, job, old_job, Time.now.to_f - @failed_job_lifespan, @failed_job_max_count] ) end end
pick()
click to toggle source
# File lib/pallets/backends/redis.rb, line 27 def pick @pool.execute do |client| job = client.brpoplpush(QUEUE_KEY, RELIABILITY_QUEUE_KEY, timeout: @blocking_timeout) if job # We store the job's timeout so we know when to retry jobs that are # still on the reliability queue. We do this separately since there is # no other way to atomically BRPOPLPUSH from the main queue to a # sorted set client.zadd(RELIABILITY_SET_KEY, Time.now.to_f + @job_timeout, job) end job end end
reschedule_all(earlier_than)
click to toggle source
# File lib/pallets/backends/redis.rb, line 87 def reschedule_all(earlier_than) @pool.execute do |client| client.evalsha( @scripts['reschedule_all'], [RELIABILITY_SET_KEY, RELIABILITY_QUEUE_KEY, RETRY_SET_KEY, QUEUE_KEY], [earlier_than] ) end end
retry(job, old_job, at)
click to toggle source
# File lib/pallets/backends/redis.rb, line 57 def retry(job, old_job, at) @pool.execute do |client| client.evalsha( @scripts['retry'], [RETRY_SET_KEY, RELIABILITY_QUEUE_KEY, RELIABILITY_SET_KEY], [at, job, old_job] ) end end
run_workflow(wfid, jobs, jobmasks, context_buffer)
click to toggle source
# File lib/pallets/backends/redis.rb, line 97 def run_workflow(wfid, jobs, jobmasks, context_buffer) @pool.execute do |client| client.multi do jobmasks.each { |jid, jobmask| client.zadd(JOBMASK_KEY % jid, jobmask) } client.sadd(JOBMASKS_KEY % wfid, jobmasks.map { |jid, _| JOBMASK_KEY % jid }) unless jobmasks.empty? client.evalsha( @scripts['run_workflow'], [WORKFLOW_QUEUE_KEY % wfid, QUEUE_KEY, REMAINING_KEY % wfid], jobs ) client.hmset(CONTEXT_KEY % wfid, *context_buffer.to_a) unless context_buffer.empty? end end end
save(wfid, jid, job, context_buffer)
click to toggle source
# File lib/pallets/backends/redis.rb, line 47 def save(wfid, jid, job, context_buffer) @pool.execute do |client| client.evalsha( @scripts['save'], [WORKFLOW_QUEUE_KEY % wfid, QUEUE_KEY, RELIABILITY_QUEUE_KEY, RELIABILITY_SET_KEY, CONTEXT_KEY % wfid, REMAINING_KEY % wfid, JOBMASK_KEY % jid, JOBMASKS_KEY % wfid], context_buffer.to_a << job ) end end
Private Instance Methods
register_scripts()
click to toggle source
# File lib/pallets/backends/redis.rb, line 114 def register_scripts @scripts ||= @pool.execute do |client| Dir["#{__dir__}/scripts/*.lua"].map do |file| name = File.basename(file, '.lua') script = File.read(file) sha = client.script(:load, script) [name, sha] end.to_h end end