Class: DynportTools::RedisQ

Inherits:
Object
  • Object
show all
Defined in:
lib/dynport_tools/redis_q.rb

Constant Summary collapse

DEFAULTS = 
{ :retry_count => 3 }

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(redis_key, options = {}) ⇒ RedisQ

Returns a new instance of RedisQ.



5
6
7
8
9
 
# File 'lib/dynport_tools/redis_q.rb', line 5

def initialize(redis_key, options = {})
  DEFAULTS.merge(options).merge(:redis_key => redis_key).each do |key, value|
    self.send(:"#{key}=", value) if self.respond_to?(:"#{key}=")
  end
end

Instance Attribute Details

#redis ⇒ Object

Returns the value of attribute redis.



3
4
5
 
# File 'lib/dynport_tools/redis_q.rb', line 3

def redis
  @redis
end

#redis_key ⇒ Object

Returns the value of attribute redis_key.



3
4
5
 
# File 'lib/dynport_tools/redis_q.rb', line 3

def redis_key
  @redis_key
end

#retry_count ⇒ Object

Returns the value of attribute retry_count.



3
4
5
 
# File 'lib/dynport_tools/redis_q.rb', line 3

def retry_count
  @retry_count
end

Instance Method Details

#count ⇒ Object 



39
40
41
 
# File 'lib/dynport_tools/redis_q.rb', line 39

def count
  redis.zcard(redis_key)
end

#each(options = {}) ⇒ Object 



54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
 
# File 'lib/dynport_tools/redis_q.rb', line 54

def each(options = {})
  entries_with_errors = []
  stats = { :errors => {}, :ok => [] }
  batch_size = options[:batch_size].to_i if options[:batch_size].to_i > 0
  while (result_hash = pop(batch_size || 1)).any?
    begin
      yield(batch_size ? result_hash.to_a.sort_by { |a| a.last }.reverse.map { |a| a.first } : result_hash.keys.first)
      stats[:ok] << result_hash.keys.join(",")
    rescue => err
      stats[:errors][result_hash.keys.join(",")] = ([err.message] + err.backtrace[0,5]).join("\n")
      entries_with_errors << result_hash if mark_failed(result_hash.keys.join(",")) < retry_count
    end
  end
  push_many(entries_with_errors, :failed => true) if entries_with_errors.any?
  stats
end

#failed_key ⇒ Object 



75
76
77
 
# File 'lib/dynport_tools/redis_q.rb', line 75

def failed_key
  "#{redis_key}/failed_counts"
end

#failed_tries ⇒ Object 



50
51
52
 
# File 'lib/dynport_tools/redis_q.rb', line 50

def failed_tries
  @failed_tries ||= Hash.new(0)
end

#mark_failed(id) ⇒ Object 



71
72
73
 
# File 'lib/dynport_tools/redis_q.rb', line 71

def mark_failed(id)
  redis.zincrby(failed_key, 1, id).to_i
end

#nil_or_lower?(a, b) ⇒ Boolean

Returns:

  • (Boolean) 


31
32
33
 
# File 'lib/dynport_tools/redis_q.rb', line 31

def nil_or_lower?(a, b)
  a.nil? || a.to_i < b
end

#pop(number = 1) ⇒ Object 



43
44
45
46
47
48
 
# File 'lib/dynport_tools/redis_q.rb', line 43

def pop(number = 1)
  Hash[*redis.multi do
    redis.zrevrange(redis_key, 0, number - 1, :with_scores => true)
    redis.zremrangebyrank(redis_key, 0 - number, -1)
  end.first]
end

#priority_of(id) ⇒ Object 



35
36
37
 
# File 'lib/dynport_tools/redis_q.rb', line 35

def priority_of(id)
  redis.zscore(redis_key, id)
end

#push(id, priority = nil, options = {}) ⇒ Object 



11
12
13
14
15
16
17
18
19
 
# File 'lib/dynport_tools/redis_q.rb', line 11

def push(id, priority = nil, options = {})
  priority ||= Time.now.to_i * -1
  if nil_or_lower?(priority_of(id), priority)
    redis.multi if !options[:no_multi]
    redis.zrem(failed_key, id) if !options[:failed]
    redis.zadd(redis_key, priority, id)
    redis.exec if !options[:no_multi]
  end
end

#push_many(array, options = {}) ⇒ Object 



21
22
23
24
25
26
27
28
29
 
# File 'lib/dynport_tools/redis_q.rb', line 21

def push_many(array, options = {})
  redis.multi do
    array.each do | (id, popularity) |
      (id.is_a?(Hash) ? id : { id => popularity }).each do |id2, popularity2|
        push(id2, popularity2, options.merge(:no_multi => true))
      end
    end
  end
end