class Resque::Plugins::Brokered::Broker
Public Class Methods
new(redis, queues)
click to toggle source
# File lib/resque/plugins/brokered/broker.rb, line 5 def initialize redis, queues @redis = redis @queues = queues end
Public Instance Methods
available_queues()
click to toggle source
# File lib/resque/plugins/brokered/broker.rb, line 10 def available_queues @redis.sdiff :queues, :active_queues end
filter_queues(queues)
click to toggle source
# File lib/resque/plugins/brokered/broker.rb, line 14 def filter_queues queues queues.select {|q| queues_regex.match q} end
get_queue()
click to toggle source
# File lib/resque/plugins/brokered/broker.rb, line 18 def get_queue queues = filter_queues(available_queues) queues.shuffle.detect {|name| @redis.llen("queue:#{name}") > 0 } end
pop()
click to toggle source
# File lib/resque/plugins/brokered/broker.rb, line 27 def pop @redis.watch "#{@redis.namespace}:active_queues" return nil unless queue_name = get_queue @redis.multi @redis.sadd :active_queues, queue_name @redis.lpop "queue:#{queue_name}" add, value = @redis.exec [queue_name, Resque.decode(value)] ensure @redis.unwatch end
queues_regex()
click to toggle source
# File lib/resque/plugins/brokered/broker.rb, line 23 def queues_regex /^(?:#{@queues.join('|')}).*/ end