Class: Derrick::Aggregator

Inherits:
Object
  • Object
show all
Defined in:
lib/derrick/aggregator.rb

Constant Summary collapse

ANY = 
'*'.freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(queue, context) ⇒ Aggregator

Returns a new instance of Aggregator.



6
7
8
9
10
 
# File 'lib/derrick/aggregator.rb', line 6

def initialize(queue, context)
  @queue = queue
  @patterns = {}
  @context = context
end

Instance Attribute Details

#patterns ⇒ Object (readonly)

Returns the value of attribute patterns.



5
6
7
 
# File 'lib/derrick/aggregator.rb', line 5

def patterns
  @patterns
end

Instance Method Details

#aggregate(key) ⇒ Object 



26
27
28
29
30
31
32
33
 
# File 'lib/derrick/aggregator.rb', line 26

def aggregate(key)
  pattern = pattern_from(key)
  pattern.aggregate(key)

  if patterns.size > @context.max_patterns
    compact_uniques!
  end
end

#compact_uniques! ⇒ Object 



39
40
41
42
43
44
45
46
47
48
 
# File 'lib/derrick/aggregator.rb', line 39

def compact_uniques!
  any = @patterns.delete(ANY) || Pattern.new
  @patterns.each do |key, aggregate|
    if aggregate.count == 1
      any.merge!(@patterns.delete(key))
    end
  end
  @patterns[ANY] = any
  nil
end

#pattern_from(key) ⇒ Object 



35
36
37
 
# File 'lib/derrick/aggregator.rb', line 35

def pattern_from(key)
  @patterns[Pattern.extract(key.name)] ||= Pattern.new
end

#run ⇒ Object 



12
13
14
15
16
17
18
19
20
21
22
23
24
 
# File 'lib/derrick/aggregator.rb', line 12

def run
  fetcher_count = @context.concurrency
  loop do
    keys = @queue.pop
    if keys == :stop
      fetcher_count -= 1
      break if fetcher_count == 0
    else
      keys.each { |k| aggregate(k) }
    end
  end
  self
end