Class: Microphite::Client::Base

Inherits:
Object
  • Object
show all
Defined in:
lib/microphite/client/base.rb

Direct Known Subclasses

Noop, Private::Prefixed, Socket

Instance Method Summary collapse

Constructor Details

#initialize(options) ⇒ Base

Returns a new instance of Base.



7
8
9
10
11
12
13
14
15
16
17
18
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
 
# File 'lib/microphite/client/base.rb', line 7

def initialize(options)
  defaults = {
      prefix: '',            # Prefix every key with this
      flush_interval: 1.0,   # How often to flush gathered metrics
      limit: 1000000,        # Size limit for gather and write stacks
      error_handler: nil     # Callback block for exceptions (mainly for debug/dev)
  }
  params = defaults.merge(options)

  # Read-only state
  @prefix           = params[:prefix]
  @limit            = params[:limit]
  @flush_interval   = params[:flush_interval]
  @error_handler    = params[:error_handler]

  # Shared state
  @status           = :running
  @gather_queue     = Queue.new
  @write_queue      = Queue.new

  # Worker state
  @accumulating     = {}
  @next_flush       = now + @flush_interval

  # Synchronization primitives
  @lock             = Mutex.new
  @worker_event     = ConditionVariable.new   # Signals worker to wake-up
  @shutdown_event   = ConditionVariable.new   # Signals close() caller when shutdown is complete

  # The worker thread does all of the data processing and socket writing
  Thread.new do
    wrap_errors do
      startup

      # If startup throws, then worker_loop will never be called (what we want)
      worker_loop
    end
  end
end

Instance Method Details

#close(timeout = nil) ⇒ Object 



76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
 
# File 'lib/microphite/client/base.rb', line 76

def close(timeout=nil)
  @lock.synchronize do
    case @status
      when :running, :ending
        @status = :ending
        @worker_event.signal
        @shutdown_event.wait(@lock, timeout)
        return @status == :shutdown
      when :shutdown
        return true
      else
        error(AssertionError.new("Invalid status: #{@status}"))
    end
  end
end

#gather(metrics) ⇒ Object 



52
53
54
55
 
# File 'lib/microphite/client/base.rb', line 52

def gather(metrics)
  return false unless metrics.is_a? Hash
  push(@gather_queue, metrics)
end

#prefix(prefix, &block) ⇒ Object 



57
58
59
60
61
62
63
 
# File 'lib/microphite/client/base.rb', line 57

def prefix(prefix, &block)
  prefixed = Private::Prefixed.new(self, prefix)
  if block_given?
    block.call(prefixed)
  end
  prefixed
end

#time(key, &block) ⇒ Object 



65
66
67
68
69
70
71
72
73
74
 
# File 'lib/microphite/client/base.rb', line 65

def time(key, &block)
  if block_given?
    before = now
    result = block.call
    after = now
    elapsed = after - before
    gather(key => elapsed)
    result
  end
end

#write(metrics) ⇒ Object 



47
48
49
50
 
# File 'lib/microphite/client/base.rb', line 47

def write(metrics)
  return false unless metrics.is_a? Hash
  push(@write_queue, metrics)
end