use the resque queue class to power the Rescue class methods

Author: tenderlove

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

@@ -4,18 +4,24 @@
 require 'mutex_m'
 
 module Resque
+  ###
+  # A queue interface that quacks like Queue from Ruby's stdlib.
   class Queue
     include Mutex_m
 
     VERSION = '1.0.0'
 
+    ###
+    # 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)
@@ -25,30 +31,59 @@ def push object
     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.rpop(@name)
+          value = @redis.lpop(@name)
           raise ThreadError unless value
           decode value
         end
       else
         synchronize do
-          value = @redis.brpop(@name, 1) until value
+          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