add queue code and specs

Author: edk

Gemfile

@@ -2,5 +2,9 @@ source 'https://rubygems.org'
 
 gem 'rake'
 # Specify your gem's dependencies in resque-round-robin.gemspec
+gem 'linecache19', '0.5.13', :path => "~/.rvm/gems/ruby-1.9.3-p0/gems/linecache19-0.5.13/"
+gem 'ruby-debug-base19', '0.11.26', :path => "~/.rvm/gems/ruby-1.9.3-p0/gems/ruby-debug-base19-0.11.26/"
+gem 'ruby-debug19', :require => 'ruby-debug'
+
 gemspec
 

README.md

@@ -11,10 +11,7 @@ For our situation, which is probably pretty rare in rails deployments,
 we have multiple customers who submit jobs to resque, and we need to
 keep the jobs of one customer from starving out other customers.
 
-Using resque-dynamic-queues in conjunction with this gem, we are able to
-support this requirement of fairness, where not more than one worker will
-process each customers job.
-
+resque-dynamic-queues is a pre-requisite.
 
 ## Installation
 

lib/resque-round-robin.rb

@@ -1,4 +1,8 @@
 
 require 'resque'
-require "resque/plugins/round-robin/version"
+require 'resque/worker'
+require 'resque-dynamic-queues'
+require "resque/plugins/round_robin/version"
+require "resque/plugins/round_robin/round_robin"
 
+Resque::Worker.send(:include, Resque::Plugins::RoundRobin)

lib/resque/plugins/round_robin/round_robin.rb

@@ -1,10 +1,43 @@
 
-module Resque
-  module Plugins
-    module RoundRobin
-      
+module Resque::Plugins
+  module RoundRobin
+    def filter_busy_queues qs
+      busy_queues = Resque::Worker.working.map { |worker| worker.job["queue"] }.compact
+      Array(qs.dup).compact - busy_queues
     end
-  end
-end
 
+    def rotated_queues
+      @n ||= 0
+      @n += 1
+      rot_queues = queues
+      rot_queues.rotate(@n % rot_queues.size)
+    end
+
+    def reserve_with_round_robin
+      qs = rotated_queues
+      qs = filter_busy_queues qs
+      qs.each do |queue|
+        log! "Checking #{queue}"
+        if job = Resque::Job.reserve(queue)
+          log! "Found job on #{queue}"
+          return job
+        end
+      end
+
+      nil
+    rescue Exception => e
+      log "Error reserving job: #{e.inspect}"
+      log e.backtrace.join("\n")
+      raise e
+    end
+
+    def self.included(receiver)
+      receiver.class_eval do
+        alias reserve_without_round_robin reserve
+        alias reserve reserve_with_round_robin
+      end
+    end
+
+  end # RoundRobin
+end # Resque::Plugins
 

resque-round-robin.gemspec

@@ -9,6 +9,10 @@ Gem::Specification.new do |gem|
   gem.homepage      = ""
 
   gem.add_dependency "resque"
+  gem.add_dependency "resque-dynamic-queues"
+
+  gem.add_development_dependency('rspec', '~> 2.5')
+  gem.add_development_dependency('rack-test', '~> 0.5.4')
 
   gem.executables   = `git ls-files -- bin/*`.split("\n").map{ |f| File.basename(f) }
   gem.files         = `git ls-files`.split("\n")