Client workers can now define associated queue

Configure client middleware chain by default
Middleware entries should be unique
Change client#push to return boolean based on pushed or not.

lib/sidekiq/client.rb

@@ -8,40 +8,46 @@ require 'sidekiq/middleware/client/unique_jobs'
 
 module Sidekiq
   class Client
+
     def self.middleware
-      @middleware ||= Middleware::Chain.new
+      @middleware ||= begin
+        m = Middleware::Chain.new
+        m.register do
+          use Middleware::Client::UniqueJobs, Client.redis
+          use Middleware::Client::ResqueWebCompatibility, Client.redis
+        end
+        m
+      end
+    end
+
+    def self.queues
+      @queues ||= {}
     end
 
     def self.redis
-      @redis ||= begin
+      @redis ||= RedisConnection.create
-        RedisConnection.create
-      end
     end
 
     def self.redis=(redis)
       @redis = redis
     end
 
-    def self.ignore_duplicate_jobs=(ignore)
-      if ignore
-        middleware.register do
-          use Middleware::Client::UniqueJobs, Client.redis
-        end
-      else
-        middleware.unregister(Middleware::Client::UniqueJobs)
-      end
-    end
-
     # Example usage:
     # Sidekiq::Client.push('my_queue', 'class' => MyWorker, 'args' => ['foo', 1, :bat => 'bar'])
-    def self.push(queue='default', item)
+    def self.push(queue=nil, item)
       raise(ArgumentError, "Message must be a Hash of the form: { 'class' => SomeClass, 'args' => ['bob', 1, :foo => 'bar'] }") unless item.is_a?(Hash)
       raise(ArgumentError, "Message must include a class and set of arguments: #{item.inspect}") if !item['class'] || !item['args']
 
+      queue = queue || queues[item['class'].to_s] || 'default'
+
       item['class'] = item['class'].to_s if !item['class'].is_a?(String)
+
+      pushed = false
       middleware.invoke(item, queue) do
         redis.rpush("queue:#{queue}", MultiJson.encode(item))
+        pushed = true
       end
+      pushed
     end
 
     # Please use .push if possible instead.
@@ -53,7 +59,7 @@ module Sidekiq
     # Messages are enqueued to the 'default' queue.
     #
     def self.enqueue(klass, *args)
-      push('default', { 'class' => klass.name, 'args' => args })
+      push(nil, { 'class' => klass.name, 'args' => args })
     end
   end
 end

lib/sidekiq/middleware/chain.rb

@@ -53,7 +53,11 @@ module Sidekiq
       end
 
       def use(klass, *args)
-        entries << Entry.new(klass, *args)
+        entries << Entry.new(klass, *args) unless exists?(klass)
+      end
+
+      def exists?(klass)
+        entries.any? { |entry| entry.klass == klass }
       end
 
       def retrieve

lib/sidekiq/worker.rb

@@ -37,6 +37,10 @@ module Sidekiq
       def perform_async(*args)
         Sidekiq::Client.push('class' => self.name, 'args' => args)
       end
+
+      def queue(name)
+        Sidekiq::Client.queues[self.name] = name.to_s
+      end
     end
   end
 end

test/test_client.rb

@@ -10,13 +10,17 @@ class TestClient < MiniTest::Unit::TestCase
     end
 
     it 'does not push duplicate messages when configured for unique only' do
-      Sidekiq::Client.ignore_duplicate_jobs = true
+      Sidekiq::Client.middleware.entries.clear
+      Sidekiq::Client.middleware.register do
+        use Sidekiq::Middleware::Client::UniqueJobs, Sidekiq::Client.redis
+        use Sidekiq::Middleware::Client::ResqueWebCompatibility, Sidekiq::Client.redis
+      end
       10.times { Sidekiq::Client.push('customqueue', 'class' => 'Foo', 'args' => [1, 2]) }
       assert_equal 1, Sidekiq::Client.redis.llen("queue:customqueue")
     end
 
     it 'does push duplicate messages when not configured for unique only' do
-      Sidekiq::Client.ignore_duplicate_jobs = false
+      Sidekiq::Client.middleware.unregister(Sidekiq::Middleware::Client::UniqueJobs)
       10.times { Sidekiq::Client.push('customqueue2', 'class' => 'Foo', 'args' => [1, 2]) }
       assert_equal 10, Sidekiq::Client.redis.llen("queue:customqueue2")
     end
@@ -27,9 +31,10 @@ class TestClient < MiniTest::Unit::TestCase
       @redis = MiniTest::Mock.new
       def @redis.multi; yield; end
       def @redis.set(*); true; end
+      def @redis.sadd(*); true; end
+      def @redis.get(*); nil; end
       def @redis.expire(*); true; end
       Sidekiq::Client.redis = @redis
-      Sidekiq::Client.ignore_duplicate_jobs = false
     end
 
     it 'raises ArgumentError with invalid params' do
@@ -44,8 +49,8 @@ class TestClient < MiniTest::Unit::TestCase
 
     it 'pushes messages to redis' do
       @redis.expect :rpush, 1, ['queue:foo', String]
-      count = Sidekiq::Client.push('foo', 'class' => 'Foo', 'args' => [1, 2])
+      pushed = Sidekiq::Client.push('foo', 'class' => 'Foo', 'args' => [1, 2])
-      assert count > 0
+      assert pushed
       @redis.verify
     end
 
@@ -55,15 +60,28 @@ class TestClient < MiniTest::Unit::TestCase
 
     it 'handles perform_async' do
       @redis.expect :rpush, 1, ['queue:default', String]
-      count = MyWorker.perform_async(1, 2)
+      pushed = MyWorker.perform_async(1, 2)
-      assert count > 0
+      assert pushed
       @redis.verify
     end
 
     it 'enqueues messages to redis' do
       @redis.expect :rpush, 1, ['queue:default', String]
-      count = Sidekiq::Client.enqueue(MyWorker, 1, 2)
+      pushed = Sidekiq::Client.enqueue(MyWorker, 1, 2)
-      assert count > 0
+      assert pushed
+      @redis.verify
+    end
+
+    class QueuedWorker
+      include Sidekiq::Worker
+
+      queue :flimflam
+    end
+
+    it 'enqueues to the named queue' do
+      @redis.expect :rpush, 1, ['queue:flimflam', String]
+      pushed = QueuedWorker.perform_async(1, 2)
+      assert pushed
       @redis.verify
     end
   end

test/test_manager.rb

@@ -26,7 +26,6 @@ class TestManager < MiniTest::Unit::TestCase
     end
 
     it 'processes messages' do
-      Sidekiq::Client.ignore_duplicate_jobs = false
       Sidekiq::Client.push(:foo, 'class' => IntegrationWorker, 'args' => [1, 2])
       Sidekiq::Client.push(:foo, 'class' => IntegrationWorker, 'args' => [1, 2])
 

test/test_middleware.rb

@@ -55,7 +55,7 @@ class TestMiddleware < MiniTest::Unit::TestCase
       processor = Sidekiq::Processor.new(boss)
       boss.expect(:processor_done!, nil, [processor])
       processor.process(msg)
-      assert_equal %w(0 before 1 before work_performed 1 after 0 after), recorder.flatten
+      assert_equal %w(0 before work_performed 0 after), recorder.flatten
     end
 
     it 'allows middleware to abruptly stop processing rest of chain' do