Class: Qu::Backend::Redis

Inherits:
Base
  • Object
show all
Defined in:
lib/qu/backend/redis.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Redis

Returns a new instance of Redis.



9
10
11
 
# File 'lib/qu/backend/redis.rb', line 9

def initialize
  self.namespace = :qu
end

Instance Attribute Details

#namespace ⇒ Object

Returns the value of attribute namespace.



7
8
9
 
# File 'lib/qu/backend/redis.rb', line 7

def namespace
  @namespace
end

Instance Method Details

#clear(queue = nil) ⇒ Object 



31
32
33
34
35
36
37
38
39
 
# File 'lib/qu/backend/redis.rb', line 31

def clear(queue = nil)
  queue ||= queues + ['failed']
  logger.info { "Clearing queues: #{queue.inspect}" }
  Array(queue).each do |q|
    logger.debug "Clearing queue #{q}"
    redis.srem('queues', q)
    redis.del("queue:#{q}")
  end
end

#clear_workers ⇒ Object 



98
99
100
101
102
103
104
 
# File 'lib/qu/backend/redis.rb', line 98

def clear_workers
  logger.info "Clearing workers"
  while id = redis.spop(:workers)
    logger.debug "Clearing worker #{id}"
    redis.del("worker:#{id}")
  end
end

#completed(payload) ⇒ Object 



67
68
69
 
# File 'lib/qu/backend/redis.rb', line 67

def completed(payload)
  redis.del("job:#{payload.id}")
end

#connection ⇒ Object Also known as: redis 



13
14
15
 
# File 'lib/qu/backend/redis.rb', line 13

def connection
  @connection ||= ::Redis::Namespace.new(namespace, :redis => ::Redis.connect(:url => ENV['REDISTOGO_URL']))
end

#enqueue(payload) ⇒ Object 



18
19
20
21
22
23
24
25
 
# File 'lib/qu/backend/redis.rb', line 18

def enqueue(payload)
  payload.id = SimpleUUID::UUID.new.to_guid
  redis.set("job:#{payload.id}", encode('klass' => payload.klass.to_s, 'args' => payload.args))
  redis.rpush("queue:#{payload.queue}", payload.id)
  redis.sadd('queues', payload.queue)
  logger.debug { "Enqueued job #{payload}" }
  payload
end

#failed(payload, error) ⇒ Object 



63
64
65
 
# File 'lib/qu/backend/redis.rb', line 63

def failed(payload, error)
  redis.rpush("queue:failed", payload.id)
end

#length(queue = 'default') ⇒ Object 



27
28
29
 
# File 'lib/qu/backend/redis.rb', line 27

def length(queue = 'default')
  redis.llen("queue:#{queue}")
end

#queues ⇒ Object 



41
42
43
 
# File 'lib/qu/backend/redis.rb', line 41

def queues
  Array(redis.smembers('queues'))
end

#register_worker(worker) ⇒ Object 



82
83
84
85
86
 
# File 'lib/qu/backend/redis.rb', line 82

def register_worker(worker)
  logger.debug "Registering worker #{worker.id}"
  redis.set("worker:#{worker.id}", encode(worker.attributes))
  redis.sadd(:workers, worker.id)
end

#release(payload) ⇒ Object 



59
60
61
 
# File 'lib/qu/backend/redis.rb', line 59

def release(payload)
  redis.rpush("queue:#{payload.queue}", payload.id)
end

#requeue(id) ⇒ Object 



71
72
73
74
75
76
77
78
79
80
 
# File 'lib/qu/backend/redis.rb', line 71

def requeue(id)
  logger.debug "Requeuing job #{id}"
  if payload = get(id)
    redis.lrem('queue:failed', 1, id)
    redis.rpush("queue:#{payload.queue}", id)
    payload
  else
    false
  end
end

#reserve(worker, options = {:block => true}) ⇒ Object 



45
46
47
48
49
50
51
52
53
54
55
56
57
 
# File 'lib/qu/backend/redis.rb', line 45

def reserve(worker, options = {:block => true})
  queues = worker.queues.map {|q| "queue:#{q}" }

  logger.debug { "Reserving job in queues #{queues.inspect}"}

  if options[:block]
    id = redis.blpop(*queues.push(0))[1]
  else
    queues.detect {|queue| id = redis.lpop(queue) }
  end

  get(id) if id
end

#unregister_worker(worker) ⇒ Object 



88
89
90
91
92
 
# File 'lib/qu/backend/redis.rb', line 88

def unregister_worker(worker)
  logger.debug "Unregistering worker #{worker.id}"
  redis.del("worker:#{worker.id}")
  redis.srem('workers', worker.id)
end

#workers ⇒ Object 



94
95
96
 
# File 'lib/qu/backend/redis.rb', line 94

def workers
  Array(redis.smembers(:workers)).map { |id| worker(id) }.compact
end