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