Merge pull request resque#561 from tenderlove/queue

Implement a Resque::Queue class

Author: hone

lib/resque.rb

@@ -12,6 +12,7 @@
 require 'resque/job'
 require 'resque/worker'
 require 'resque/plugin'
+require 'resque/queue'
 
 module Resque
   include Helpers
@@ -43,6 +44,9 @@ def redis=(server)
     else
       @redis = Redis::Namespace.new(:resque, :redis => server)
     end
+    @queues = Hash.new { |h,name|
+      h[name] = Resque::Queue.new(name, @redis, self)
+    }
   end
 
   # Returns the current Redis connection. If none has been created, will
@@ -137,20 +141,24 @@ def to_s
   # Returns nothing
   def push(queue, item)
     watch_queue(queue)
-    redis.rpush "queue:#{queue}", encode(item)
+    @queues[queue] << item
   end
 
   # Pops a job off a queue. Queue name should be a string.
   #
   # Returns a Ruby object.
   def pop(queue)
-    decode redis.lpop("queue:#{queue}")
+    begin
+      @queues[queue].pop(true)
+    rescue ThreadError
+      nil
+    end
   end
 
   # Returns an integer representing the size of a queue.
   # Queue name should be a string.
   def size(queue)
-    redis.llen("queue:#{queue}").to_i
+    @queues[queue].size
   end
 
   # Returns an array of items currently queued. Queue name should be
@@ -162,7 +170,7 @@ def size(queue)
   # To get the 3rd page of a 30 item, paginatied list one would use:
   #   Resque.peek('my_list', 59, 30)
   def peek(queue, start = 0, count = 1)
-    list_range("queue:#{queue}", start, count)
+    @queues[queue].slice start, count
   end
 
   # Does the dirty work of fetching a range of items from a Redis list
@@ -185,7 +193,7 @@ def queues
   # Given a queue name, completely deletes the queue.
   def remove_queue(queue)
     redis.srem(:queues, queue.to_s)
-    redis.del("queue:#{queue}")
+    @queues[queue].destroy
   end
 
   # Used internally to keep track of which queues we've created.

lib/resque/helpers.rb

@@ -21,6 +21,7 @@ def redis
     def encode(object)
       ::MultiJson.encode(object)
     end
+    alias :dump :encode
 
     # Given a string, returns a Ruby object.
     def decode(object)
@@ -32,6 +33,7 @@ def decode(object)
         raise DecodeException, e.message, e.backtrace
       end
     end
+    alias :load :decode
 
     # Given a word with dashes, returns a camel cased version of it.
     #

lib/resque/queue.rb

@@ -0,0 +1,94 @@
+require 'redis'
+require 'redis-namespace'
+require 'thread'
+require 'mutex_m'
+
+module Resque
+  ###
+  # A queue interface that quacks like Queue from Ruby's stdlib.
+  class Queue
+    include Mutex_m
+
+    ###
+    # 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, coder = Marshal
+      super()
+      @name  = "queue:#{name}"
+      @redis = redis
+      @coder = coder
+    end
+
+    # Add +object+ to the queue
+    def push object
+      synchronize do
+        @redis.rpush @name, encode(object)
+      end
+    end
+
+    alias :<< :push
+    alias :enq :push
+
+    # Returns a list of objects in the queue.  This method is *not* available
+    # on the stdlib Queue.
+    def slice start, length
+      if length == 1
+        synchronize do
+          decode @redis.lindex @name, start
+        end
+      else
+        synchronize do
+          Array(@redis.lrange(@name, start, start + length - 1)).map do |item|
+            decode item
+          end
+        end
+      end
+    end
+
+    # Pop an item off the queue.  This method will block until an item is
+    # available.
+    #
+    # Pass +true+ for a non-blocking pop.  If nothing is read on a non-blocking
+    # pop, a ThreadError is raised.
+    def pop non_block = false
+      if non_block
+        synchronize do
+          value = @redis.lpop(@name)
+          raise ThreadError unless value
+          decode value
+        end
+      else
+        synchronize do
+          value = @redis.blpop(@name, 1) until value
+          decode value.last
+        end
+      end
+    end
+
+    # Get the length of the queue
+    def length
+      @redis.llen @name
+    end
+    alias :size :length
+
+    # Is the queue empty?
+    def empty?
+      size == 0
+    end
+
+    # Deletes this Queue from redis. This method is *not* available on the
+    # stdlib Queue.
+    def destroy
+      @redis.del @name
+    end
+
+    private
+    def encode object
+      @coder.dump object
+    end
+
+    def decode object
+      @coder.load object
+    end
+  end
+end

test/redis_queue_test.rb

@@ -0,0 +1,80 @@
+require 'test_helper'
+require 'resque/queue'
+
+module Resque
+  class TestQueue < MiniTest::Unit::TestCase
+    class Thing
+      attr_reader :inside
+
+      def initialize
+        @inside = "x"
+      end
+
+      def == other
+        super || @inside == other.inside
+      end
+    end
+
+    def test_sanity
+      queue = q
+      x = Thing.new
+      queue.push x
+      assert_equal x, queue.pop
+    end
+
+    def test_pop_blocks
+      queue1 = q
+      queue2 = q
+
+      t = Thread.new { queue1.pop }
+      x = Thing.new
+
+      queue2.push x
+      assert_equal x, t.join.value
+    end
+
+    def test_nonblock_pop
+      queue1 = q
+
+      assert_raises ThreadError do
+        queue1.pop(true)
+      end
+    end
+
+    def test_pop_blocks_forever
+      queue1 = q
+      assert_raises Timeout::Error do
+        Timeout.timeout(2) { queue1.pop }
+      end
+    end
+
+    def test_size
+      queue = q
+      assert_equal 0, queue.size
+
+      queue << Thing.new
+      assert_equal 1, queue.size
+    ensure
+      queue.pop
+    end
+
+    def test_empty?
+      queue = q
+      assert queue.empty?
+
+      queue << Thing.new
+      refute queue.empty?
+    ensure
+      queue.pop
+    end
+
+    def q
+      Queue.new 'foo', backend
+    end
+
+    def backend
+      redis = Redis.new(:host => "127.0.0.1", :port => 9736)
+      Redis::Namespace.new :resque, :redis => redis
+    end
+  end
+end