Class: Resque::Queue

Inherits:
Object
  • Object
show all
Defined in:
lib/resque-priority-jobs/queue.rb

Instance Method Summary collapse

Instance Method Details

#is_a_priority_queue? ⇒ Boolean

To identify whether a queue is priority queue or not. Empty queues are always non-priority Assumption : Once a queue type is set to be a priority queue, it cannot be changed and vice-versa.

Returns:

  • (Boolean) 


61
62
63
64
 
# File 'lib/resque-priority-jobs/queue.rb', line 61

def is_a_priority_queue?
  @@queue_types ||= {}
  (@@queue_types[@redis_name] ||= queue_type) == 'zset'
end

#length ⇒ Object Also known as: size 



33
34
35
 
# File 'lib/resque-priority-jobs/queue.rb', line 33

def length
  @redis.type(@redis_name) == 'list' ? @redis.llen(@redis_name) : @redis.zcard(@redis_name)
end

#pop_with_priority(non_block = false) ⇒ Object Also known as: pop

remove entry from queue use default pop in case of non-priority queue see alias_method_chain



22
23
24
25
26
27
28
29
 
# File 'lib/resque-priority-jobs/queue.rb', line 22

def pop_with_priority non_block = false
  return pop_without_priority non_block unless is_a_priority_queue?
  synchronize do
    value = Resque::JobFetch.fetch_one_job @redis, @redis_name until non_block || value
    raise ThreadError if non_block && !value
    decode value
  end
end

#push(object) ⇒ Object Also known as: <<, enq

Raises:

  • (QueueDestroyed) 


10
11
12
13
14
15
16
 
# File 'lib/resque-priority-jobs/queue.rb', line 10

def push object
  raise QueueDestroyed if destroyed?
  return push_with_priority object['priority'].to_i, object if object['priority']
  synchronize do
    @redis.rpush @redis_name, encode(object)
  end
end

#push_with_priority(priority, object) ⇒ Object

adds entry into redis



4
5
6
7
8
 
# File 'lib/resque-priority-jobs/queue.rb', line 4

def push_with_priority priority, object
  synchronize do
    @redis.zadd @redis_name, priority, encode(object)
  end
end

#queue_type ⇒ Object 



66
67
68
69
 
# File 'lib/resque-priority-jobs/queue.rb', line 66

def queue_type
  type = @redis.type(@redis_name)
  type == 'none' ? nil : type
end

#slice_with_priority(start, length) ⇒ Object Also known as: slice 



38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
 
# File 'lib/resque-priority-jobs/queue.rb', line 38

def slice_with_priority(start, length)
  if @redis.type(@redis_name) == 'zset'
    if length == 1
      synchronize do
        @redis.zrangebyscore(@redis_name, '-inf', '+inf',  :limit =>[0,1] )
      end
    else
      synchronize do
        Array(@redis.zrangebyscore(@redis_name, '-inf', '+inf', :limit => [0, length])).map do |item|
          decode item
        end
      end
    end
  else
    slice_without_priority(start, length)
  end
end