Class: ActiveUsage::Store

Inherits:
Object
  • Object
show all
Defined in:
lib/active_usage/store.rb

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(adapter, batch_size: 100, flush_interval: 1.0, max_queue_size: 10_000) ⇒ Store

Returns a new instance of Store.



28
29
30
31
32
33
34
35
36
 
# File 'lib/active_usage/store.rb', line 28

def initialize(adapter, batch_size: 100, flush_interval: 1.0, max_queue_size: 10_000)
  @adapter = adapter
  @queue = EventQueue.new(max_queue_size, batch_size)
  @flush_mutex = Mutex.new
  @shutdown_mutex = Mutex.new
  @shutdown = false
  @worker = Worker.new(flush_interval) { flush! }
  self.class.track(self)
end

Class Method Details

.track(instance) ⇒ Object 



10
11
12
13
14
15
16
17
18
19
20
21
 
# File 'lib/active_usage/store.rb', line 10

def track(instance)
  @instances_mutex.synchronize do
    unless @exit_hook_installed
      at_exit do
        live = @instances_mutex.synchronize { @instances.dup }
        live.each(&:shutdown!)
      end
      @exit_hook_installed = true
    end
    @instances << instance
  end
end

.untrack(instance) ⇒ Object 



23
24
25
 
# File 'lib/active_usage/store.rb', line 23

def untrack(instance)
  @instances_mutex.synchronize { @instances.delete(instance) }
end

Instance Method Details

#clear! ⇒ Object 



44
45
46
47
 
# File 'lib/active_usage/store.rb', line 44

def clear!
  flush!
  @adapter.clear!
end

#flush! ⇒ Object 



49
50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/active_usage/store.rb', line 49

def flush!
  @flush_mutex.synchronize do
    batch = @queue.drain
    return if batch.empty?

    begin
      @adapter.record(batch)
    rescue StandardError
      batch.each { |event| @queue.push(event) }
      raise
    end
  end
end

#record(event) ⇒ Object 



38
39
40
41
42
 
# File 'lib/active_usage/store.rb', line 38

def record(event)
  @queue.push(event)
  flush! if @queue.flush_ready?
  event
end

#shutdown! ⇒ Object 



63
64
65
66
67
68
69
70
71
72
73
74
75
 
# File 'lib/active_usage/store.rb', line 63

def shutdown!
  @shutdown_mutex.synchronize do
    return if @shutdown

    @shutdown = true
  end

  @worker.stop!
  @worker.join(0.5)
  flush!
  @adapter.shutdown!
  self.class.untrack(self)
end