Unlock queue on dequeue

Gemfile

@@ -1,8 +1,7 @@
 source :rubygems
 
-gem "resque"
+gem "resque", :git => 'git://github.com/humancopy/resque.git'
 
 group :development do
-  gem "turn"
   gem "rake"
 end

README.md

@@ -43,4 +43,20 @@ UpdateNetworkGraph is queued at a time, regardless of the
 repo_id. Normally a job is locked using a combination of its
 class name and arguments.
 
+It is also possible to define locks which will get released
+BEFORE performing a job by overriding the lock_running? class
+method in your subclass. This is useful in cases where you need
+to get a job queued even if another job on same queue is already
+running, e.g.
+
+    class UpdateNetworkGraph
+      extend Resque::Plugins::Lock
+
+      # Do not lock a running job
+      def self.lock_running?
+        false
+      end
+    end
+
+
 [rq]: http://github.com/defunkt/resque

lib/resque/plugins/lock.rb

@@ -40,27 +40,62 @@ module Plugins
     # UpdateNetworkGraph is running at a time, regardless of the
     # repo_id. Normally a job is locked using a combination of its
     # class name and arguments.
+    # 
+    # It is also possible to define locks which will get released
+    # BEFORE performing a job by overriding the lock_running? class
+    # method in your subclass. This is useful in cases where you need
+    # to get a job queued even if another job on same queue is already
+    # running, e.g.
+    # 
+    # class UpdateNetworkGraph
+    #   extend Resque::Plugins::Lock
+    #
+    #   # Do not lock a running job
+    #   def self.lock_running?
+    #     false
+    #   end
+    # end
     module Lock
       # Override in your job to control the lock key. It is
       # passed the same arguments as `perform`, that is, your job's
       # payload.
       def lock(*args)
-        "lock:#{name}-#{args.to_s}"
+        "#{name}-#{args.to_s}"
+      end
+
+      def namespaced_lock(*args)
+        "lock:#{lock(*args)}"
       end
 
       def before_enqueue_lock(*args)
-        Resque.redis.setnx(lock(*args), true)
+        Resque.redis.setnx(namespaced_lock(*args), true)
+      end
+
+      def before_dequeue_lock(*args)
+        Resque.redis.del(namespaced_lock(*args))
+      end
+
+      def lock_running?
+        true
       end
 
       def around_perform_lock(*args)
+        before_dequeue_lock(*args) unless lock_running?
         begin
           yield
         ensure
           # Always clear the lock when we're done, even if there is an
           # error.
-          Resque.redis.del(lock(*args))
+          before_dequeue_lock(*args) if lock_running?
         end
       end
+
+      def self.all_locks
+        Resque.redis.keys('lock:*')
+      end
+      def self.clear_all_locks
+        all_locks.collect { |x| Resque.redis.del(x) }.count
+      end
     end
   end
 end

test/lock_test.rb

@@ -16,7 +16,8 @@ def self.perform
 
   def setup
     Resque.redis.del('queue:lock_test')
-    Resque.redis.del(Job.lock)
+    Resque.redis.del("lock:#{Job.lock}")
+    Resque.redis.del("lock:#{Job.lock(1)}")
   end
 
   def test_lint
@@ -37,4 +38,24 @@ def test_lock
 
     assert_equal 1, Resque.redis.llen('queue:lock_test')
   end
-end
+
+  def test_lock_with_args
+    3.times { Resque.enqueue(Job, 1) }
+
+    assert_equal 1, Resque.redis.llen('queue:lock_test')
+  end
+
+  def test_unlock
+    Resque.enqueue(Job)
+    assert_equal "true", Resque.redis.get("lock:#{Job.lock}")
+    Resque.dequeue(Job)
+    assert_nil Resque.redis.get("lock:#{Job.lock}")
+  end
+  def test_unlock_with_args
+    Resque.enqueue(Job, 1)
+    assert_equal "true", Resque.redis.get("lock:#{Job.lock(1)}")
+    Resque.dequeue(Job, 1)
+    assert_nil Resque.redis.get("lock:#{Job.lock(1)}")
+  end
+
+end