Permalink
Browse files

Use RPOPLPUSH to walk the queue in #queued. This will not have memory…

… issues when dealing with a large queue.
  • Loading branch information...
1 parent f89f205 commit d569f96a8164440f4f4a97e0e3c465d32db38c7e @jjulian jjulian committed Apr 1, 2013
Showing with 11 additions and 8 deletions.
  1. +11 −8 lib/resque/job.rb
View
19 lib/resque/job.rb
@@ -120,21 +120,24 @@ def self.destroy(queue, klass, *args)
# Whereas specifying args will only find the 2nd job:
#
# Resque::Job.queued(queue, 'UpdateGraph', 'mojombo')
- #
- # This method can be potentially very slow and memory intensive,
- # depending on the size of your queue, as it loads all jobs into
- # a Ruby array.
def self.queued(queue, klass, *args)
klass = klass.to_s
+ tmp_queue = "queue:#{queue}:tmp:#{Time.now.to_i}"
+ requeue_queue = "#{tmp_queue}:requeue"
+ jobs = []
- redis.lrange("queue:#{queue}", 0, -1).inject([]) do |memo, string|
+ while string = redis.rpoplpush("queue:#{queue}", tmp_queue)
decoded = decode(string)
if decoded['class'] == klass && (args.empty? || decoded['args'] == args)
- memo << new(queue, decoded)
+ jobs.unshift new(queue, decoded)
end
-
- memo
+ redis.rpoplpush(tmp_queue, requeue_queue)
end
+ loop do
+ redis.rpoplpush(requeue_queue, "queue:#{queue}") or break
+ end
+
+ jobs
end
# Given a string queue name, returns an instance of Resque::Job

0 comments on commit d569f96

Please sign in to comment.