Class: Resque::Failure::RedisMultiQueue

Inherits:
Base
  • Object
show all
Defined in:
lib/resque/failure/redis_multi_queue.rb

Overview

A Failure backend that stores exceptions in Redis. Very simple but works out of the box, along with support in the Resque web app.

Instance Attribute Summary

Attributes inherited from Base

#exception, #payload, #queue, #worker

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from Base

#initialize, #log, url

Constructor Details

This class inherits a constructor from Resque::Failure::Base

Class Method Details

.all(offset = 0, limit = 1, queue = :failed, order = 'desc') ⇒ Object 



36
37
38
 
# File 'lib/resque/failure/redis_multi_queue.rb', line 36

def self.all(offset = 0, limit = 1, queue = :failed, order = 'desc')
  Resque.list_range(queue, offset, limit, order)
end

.clear(queue = :failed) ⇒ Object 



57
58
59
 
# File 'lib/resque/failure/redis_multi_queue.rb', line 57

def self.clear(queue = :failed)
  Resque.redis.del(queue)
end

.count(queue = nil, class_name = nil) ⇒ Object 



20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
 
# File 'lib/resque/failure/redis_multi_queue.rb', line 20

def self.count(queue = nil, class_name = nil)
  if queue
    if class_name
      n = 0
      each(0, count(queue), queue, class_name) { n += 1 } 
      n
    else
      Resque.redis.llen(queue).to_i
    end
  else
    total = 0
    queues.each { |q| total += count(q) }
    total
  end
end

.each(offset = 0, limit = self.count, queue = :failed, class_name = nil, order = 'desc') ⇒ Object 



44
45
46
47
48
49
50
51
52
53
54
55
 
# File 'lib/resque/failure/redis_multi_queue.rb', line 44

def self.each(offset = 0, limit = self.count, queue = :failed, class_name = nil, order = 'desc')
  items = all(offset, limit, queue, order)
  items = [items] unless items.is_a? Array
  if order.eql? 'desc'
    items.reverse!
  end
  items.each_with_index do |item, i|
    if !class_name || (item['payload'] && item['payload']['class'] == class_name)
      yield offset + i, item
    end
  end
end

.queues ⇒ Object 



40
41
42
 
# File 'lib/resque/failure/redis_multi_queue.rb', line 40

def self.queues
  Array(Resque.redis.smembers(:failed_queues))
end

.remove(id, queue = :failed) ⇒ Object 



68
69
70
71
72
 
# File 'lib/resque/failure/redis_multi_queue.rb', line 68

def self.remove(id, queue = :failed)
  sentinel = ""
  Resque.redis.lset(queue, id, sentinel)
  Resque.redis.lrem(queue, 1,  sentinel)
end

.remove_queue(queue) ⇒ Object 



79
80
81
 
# File 'lib/resque/failure/redis_multi_queue.rb', line 79

def self.remove_queue(queue)
  Resque.redis.del(Resque::Failure.failure_queue_name(queue))
end

.requeue(id, queue = :failed) ⇒ Object 



61
62
63
64
65
66
 
# File 'lib/resque/failure/redis_multi_queue.rb', line 61

def self.requeue(id, queue = :failed)
  item = all(id, 1, queue)
  item['retried_at'] = Time.now.strftime("%Y/%m/%d %H:%M:%S")
  Resque.redis.lset(queue, id, Resque.encode(item))
  Job.create(item['queue'], item['payload']['class'], *item['payload']['args'])
end

.requeue_queue(queue) ⇒ Object 



74
75
76
77
 
# File 'lib/resque/failure/redis_multi_queue.rb', line 74

def self.requeue_queue(queue)
  failure_queue = Resque::Failure.failure_queue_name(queue)
  each(0, count(failure_queue), failure_queue) { |id, _| requeue(id, failure_queue) }
end

Instance Method Details

#filter_backtrace(backtrace) ⇒ Object 



83
84
85
86
 
# File 'lib/resque/failure/redis_multi_queue.rb', line 83

def filter_backtrace(backtrace)
  index = backtrace.index { |item| item.include?('/lib/resque/job.rb') }
  backtrace.first(index.to_i)
end

#save ⇒ Object 



6
7
8
9
10
11
12
13
14
15
16
17
18
 
# File 'lib/resque/failure/redis_multi_queue.rb', line 6

def save
  data = {
    :failed_at => Time.now.strftime("%Y/%m/%d %H:%M:%S %Z"),
    :payload   => payload,
    :exception => exception.class.to_s,
    :error     => UTF8Util.clean(exception.to_s),
    :backtrace => filter_backtrace(Array(exception.backtrace)),
    :worker    => worker.to_s,
    :queue     => queue
  }
  data = Resque.encode(data)
  Resque.redis.rpush(Resque::Failure.failure_queue_name(queue), data)
end