queues now use ConnectionPool

Author: hone

lib/resque/multi_queue.rb

@@ -11,12 +11,12 @@ class MultiQueue
 
     ###
     # Create a new MultiQueue using the +queues+ from the +redis+ connection
-    def initialize(queues, redis = Resque.redis)
+    def initialize(queues, pool = Resque.pool)
       super()
 
       @queues     = queues # since ruby 1.8 doesn't have Ordered Hashes
       @queue_hash = {}
-      @redis      = redis
+      @pool       = pool
 
       queues.each do |queue|
         @queue_hash[queue.redis_name] = queue
@@ -45,7 +45,7 @@ def pop(non_block = false)
       else
         queue_names = @queues.map {|queue| queue.redis_name }
         synchronize do
-          value = @redis.blpop(*(queue_names + [1])) until value
+          value = @pool.with_connection {|pool| pool.blpop(*(queue_names + [1])) } until value
           queue_name, payload = value
           queue = @queue_hash[queue_name]
           [queue, queue.decode(payload)]
@@ -59,7 +59,7 @@ def pop(non_block = false)
     # the timeout expires.
     def poll(timeout)
       queue_names = @queues.map {|queue| queue.redis_name }
-      queue_name, payload = @redis.blpop(*(queue_names + [timeout]))
+      queue_name, payload = @pool.with_connection {|pool| pool.blpop(*(queue_names + [timeout])) }
       return unless payload
 
       synchronize do

lib/resque/queue.rb

@@ -18,24 +18,26 @@ class Queue
     ###
     # Create a new Queue object with +name+ on +redis+ connection, and using
     # the +coder+ for encoding and decoding objects that are stored in redis.
-    def initialize name, redis = Resque.redis, coder = Marshal
+    def initialize name, pool = Resque.pool, coder = Marshal
       super()
       @name       = name
       @redis_name = "queue:#{@name}"
-      @redis      = redis
+      @pool       = pool
       @coder      = coder
       @destroyed  = false
 
-      @redis.sadd(:queues, @name)
+      @pool.with_connection do |conn|
+        conn.sadd(:queues, @name)
+      end
     end
 
     # Add +object+ to the queue
     # If trying to push to an already destroyed queue, it will raise a Resque::QueueDestroyed exception
     def push object
       raise QueueDestroyed if destroyed?
 
-      synchronize do
-        @redis.rpush @redis_name, encode(object)
+      @pool.with_connection do |conn|
+        conn.rpush @redis_name, synchronize {encode(object) }
       end
     end
 
@@ -47,13 +49,15 @@ def push object
     def slice start, length
       if length == 1
         synchronize do
-          decode @redis.lindex @redis_name, start
+          decode(@pool.with_connection {|conn| conn.lindex(@redis_name , start) })
         end
       else
-        synchronize do
-          Array(@redis.lrange(@redis_name, start, start + length - 1)).map do |item|
-            decode item
+        Array(
+          @pool.with_connection do |conn|
+            conn.lrange(@redis_name, start, start + length - 1)
           end
+        ).map do |item|
+          synchronize {decode item }
         end
       end
     end
@@ -65,16 +69,12 @@ def slice start, length
     # pop, a ThreadError is raised.
     def pop non_block = false
       if non_block
-        synchronize do
-          value = @redis.lpop(@redis_name)
-          raise ThreadError unless value
-          decode value
-        end
+        value = @pool.with_connection {|pool| pool.lpop(@redis_name) }
+        raise ThreadError unless value
+        synchronize {decode value }
       else
-        synchronize do
-          value = @redis.blpop(@redis_name, 1) until value
-          decode value.last
-        end
+        value = @pool.with_connection {|pool| pool.blpop(@redis_name, 1) } until value
+        synchronize {decode value.last }
       end
     end
 
@@ -83,7 +83,7 @@ def pop non_block = false
     # Blocks for +timeout+ seconds if the queue is empty, and returns nil if
     # the timeout expires.
     def poll(timeout)
-      queue_name, payload = @redis.blpop(@redis_name, timeout)
+      queue_name, payload = @pool.with_connection {|pool| pool.blpop(@redis_name, timeout) }
       return unless payload
 
       synchronize do
@@ -93,7 +93,7 @@ def poll(timeout)
 
     # Get the length of the queue
     def length
-      @redis.llen @redis_name
+      @pool.with_connection {|pool| pool.llen @redis_name }
     end
     alias :size :length
 
@@ -109,8 +109,10 @@ def empty?
     # B and you delete Queue A, pushing to Queue B will have unknown side
     # effects. Queue A will be marked destroyed, but Queue B will not.
     def destroy
-      @redis.del @redis_name
-      @redis.srem(:queues, @name)
+      @pool.with_connection do |conn|
+        conn.del @redis_name
+        conn.srem(:queues, @name)
+      end
       @destroyed = true
     end
 

lib/resque/worker.rb

@@ -204,8 +204,8 @@ def perform(job)
     def reserve(interval = 5.0)
       interval = interval.to_i
       multi_queue = MultiQueue.new(
-        queues.map {|queue| Queue.new(queue, Resque.redis, Resque.coder) },
-        Resque.redis)
+        queues.map {|queue| Queue.new(queue, Resque.pool, Resque.coder) },
+        Resque.pool)
 
       if interval < 1
         begin

test/multi_queue_test.rb

@@ -2,23 +2,24 @@
 
 describe "Resque::MulitQueue" do
   let(:redis) { Resque.redis }
+  let(:pool)  { Resque.pool }
   let(:coder) { Resque::MultiJsonCoder.new }
 
   before do
     redis.flushall
   end
 
   it "poll times out and returns nil" do
-    foo   = Resque::Queue.new 'foo', redis
-    bar   = Resque::Queue.new 'bar', redis
-    queue = Resque::MultiQueue.new([foo, bar], redis)
+    foo   = Resque::Queue.new 'foo', pool
+    bar   = Resque::Queue.new 'bar', pool
+    queue = Resque::MultiQueue.new([foo, bar], pool)
     assert_nil queue.poll(1)
   end
 
   it "blocks on pop" do
-    foo   = Resque::Queue.new 'foo', redis, coder
-    bar   = Resque::Queue.new 'bar', redis, coder
-    queue = Resque::MultiQueue.new([foo, bar], redis)
+    foo   = Resque::Queue.new 'foo', pool, coder
+    bar   = Resque::Queue.new 'bar', pool, coder
+    queue = Resque::MultiQueue.new([foo, bar], pool)
     t     = Thread.new { queue.pop }
 
     job = { 'class' => 'GoodJob', 'args' => [35, 'tar'] }
@@ -28,9 +29,9 @@
   end
 
   it "nonblocking pop works" do
-    foo   = Resque::Queue.new 'foo', redis, coder
-    bar   = Resque::Queue.new 'bar', redis, coder
-    queue = Resque::MultiQueue.new([foo, bar], redis)
+    foo   = Resque::Queue.new 'foo', pool, coder
+    bar   = Resque::Queue.new 'bar', pool, coder
+    queue = Resque::MultiQueue.new([foo, bar], pool)
 
     job = { 'class' => 'GoodJob', 'args' => [35, 'tar'] }
     bar << job
@@ -39,30 +40,30 @@
   end
 
   it "nonblocking pop doesn't block" do
-    foo   = Resque::Queue.new 'foo', redis, coder
-    bar   = Resque::Queue.new 'bar', redis, coder
-    queue = Resque::MultiQueue.new([foo, bar], redis)
+    foo   = Resque::Queue.new 'foo', pool, coder
+    bar   = Resque::Queue.new 'bar', pool, coder
+    queue = Resque::MultiQueue.new([foo, bar], pool)
 
     assert_raises ThreadError do
       queue.pop(true)
     end
   end
 
   it "blocks forever on pop" do
-    foo   = Resque::Queue.new 'foo', redis, coder
-    bar   = Resque::Queue.new 'bar', redis, coder
-    queue = Resque::MultiQueue.new([foo, bar], redis)
+    foo   = Resque::Queue.new 'foo', pool, coder
+    bar   = Resque::Queue.new 'bar', pool, coder
+    queue = Resque::MultiQueue.new([foo, bar], pool)
     assert_raises Timeout::Error do
       Timeout::timeout(2) { queue.pop }
     end
   end
 
   it "blocking pop processes queues in the order given" do
-    foo    = Resque::Queue.new 'foo', redis, coder
-    bar    = Resque::Queue.new 'bar', redis, coder
-    baz    = Resque::Queue.new 'baz', redis, coder
+    foo    = Resque::Queue.new 'foo', pool, coder
+    bar    = Resque::Queue.new 'bar', pool, coder
+    baz    = Resque::Queue.new 'baz', pool, coder
     queues = [foo, bar, baz]
-    queue  = Resque::MultiQueue.new(queues, redis)
+    queue  = Resque::MultiQueue.new(queues, pool)
     job    = { 'class' => 'GoodJob', 'args' => [35, 'tar'] }
 
     queues.each {|q| q << job }
@@ -76,11 +77,11 @@
   end
 
   it "nonblocking pop processes queues in the order given" do
-    foo    = Resque::Queue.new 'foo', redis, coder
-    bar    = Resque::Queue.new 'bar', redis, coder
-    baz    = Resque::Queue.new 'baz', redis, coder
+    foo    = Resque::Queue.new 'foo', pool, coder
+    bar    = Resque::Queue.new 'bar', pool, coder
+    baz    = Resque::Queue.new 'baz', pool, coder
     queues = [foo, bar, baz]
-    queue  = Resque::MultiQueue.new(queues, redis)
+    queue  = Resque::MultiQueue.new(queues, pool)
     job    = { 'class' => 'GoodJob', 'args' => [35, 'tar'] }
 
     queues.each {|q| q << job }

test/redis_queue_test.rb

@@ -144,10 +144,10 @@ def == other
   end
 
   def q
-    Resque::Queue.new 'foo', Resque.redis
+    Resque::Queue.new 'foo'
   end
 
   def pipe
-    [Resque::Queue.new('foo', Resque.create_connection), Resque::Queue.new('foo', Resque.create_connection)]
+    [Resque::Queue.new('foo'), Resque::Queue.new('foo')]
   end
 end

test/threaded_pool_test.rb

@@ -18,8 +18,7 @@ def run
 
     before do
       @write  = Queue.new(:foo)
-      conn = Resque.create_connection
-      @read  = Queue.new(:foo, conn)
+      @read  = Queue.new(:foo, Resque.pool)
       @tp = ThreadedPool.new(@read, 5)
     end