Class: Spanx::Actor::Collector

Inherits:
Object
  • Object
show all
Includes:
Helper::Timing
Defined in:
lib/spanx/actor/collector.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from Helper::Timing

#period_marker

Constructor Details

#initialize(config, queue) ⇒ Collector

Returns a new instance of Collector.



12
13
14
15
16
17
 
# File 'lib/spanx/actor/collector.rb', line 12

def initialize(config, queue)
  @queue = queue
  @config = config
  @semaphore = Mutex.new
  @cache = Hash.new(0)
end

Instance Attribute Details

#cache ⇒ Object

Returns the value of attribute cache.



10
11
12
 
# File 'lib/spanx/actor/collector.rb', line 10

def cache
  @cache
end

#config ⇒ Object

Returns the value of attribute config.



10
11
12
 
# File 'lib/spanx/actor/collector.rb', line 10

def config
  @config
end

#queue ⇒ Object

Returns the value of attribute queue.



10
11
12
 
# File 'lib/spanx/actor/collector.rb', line 10

def queue
  @queue
end

#semaphore ⇒ Object

Returns the value of attribute semaphore.



10
11
12
 
# File 'lib/spanx/actor/collector.rb', line 10

def semaphore
  @semaphore
end

Instance Method Details

#increment_ip(ip, timestamp) ⇒ Object 



52
53
54
 
# File 'lib/spanx/actor/collector.rb', line 52

def increment_ip(ip, timestamp)
  cache[[ip, period_marker(config[:collector][:resolution], timestamp)]] += 1
end

#run ⇒ Object 



19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
 
# File 'lib/spanx/actor/collector.rb', line 19

def run
  Thread.new do
    Thread.current[:name] = "collector:queue"
    loop do
      unless queue.empty?
        Logger.logging "caching [#{queue.size}] keys locally" do
          while !queue.empty?
            semaphore.synchronize {
              increment_ip *(queue.pop)
            }
          end
        end
      end
      sleep 1
    end
  end

  Thread.new do
    Thread.current[:name] = "collector:flush"
    loop do
      semaphore.synchronize {
        Logger.logging "flushing cache with [#{cache.keys.size}] keys" do
          cache.each_pair do |key, count|
            Spanx::IPChecker.new(key[0]).increment!(count, key[1])
          end
          reset_cache
        end
      }
      sleep config[:collector][:flush_interval]
    end
  end
end