Class: LogjamAgent::ZMQForwarder

Inherits:
Object
  • Object
show all
Includes:
Util
Defined in:
lib/logjam_agent/zmq_forwarder.rb

Constant Summary collapse

SEQUENCE_START = 
0
@@context_mutex = 
Mutex.new
@@zmq_context = 
nil

Constants included from Util

Util::BIG_ENDIAN, Util::FIXNUM_MAX, Util::META_INFO_DEVICE_NUMBER, Util::META_INFO_TAG, Util::META_INFO_VERSION, Util::UINT64

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Util

#augment_connection_spec, #next_fixnum, #pack_info, #pack_uint64_big_endian, #unpack_info, #unpack_uint64_big_endian, #zclock_time

Constructor Details

#initialize(*args) ⇒ ZMQForwarder

Returns a new instance of ZMQForwarder.



9
10
11
12
13
14
15
16
17
18
19
20
 
# File 'lib/logjam_agent/zmq_forwarder.rb', line 9

def initialize(*args)
  opts = args.extract_options!
  @app = args[0] || LogjamAgent.application_name
  @env = args[1] || LogjamAgent.environment_name
  @app_env = "#{@app}-#{@env}"
  @config = default_options.merge!(opts)
  @config[:host] = "localhost" if @config[:host].blank?
  @sequence = SEQUENCE_START
  @socket = nil
  @ping_ensured = false
  @socket_mutex = Mutex.new
end

Instance Attribute Details

#app ⇒ Object (readonly)

Returns the value of attribute app.



3
4
5
 
# File 'lib/logjam_agent/zmq_forwarder.rb', line 3

def app
  @app
end

#env ⇒ Object (readonly)

Returns the value of attribute env.



3
4
5
 
# File 'lib/logjam_agent/zmq_forwarder.rb', line 3

def env
  @env
end

Class Method Details

.context ⇒ Object 



42
43
44
45
46
47
48
49
50
51
52
53
 
# File 'lib/logjam_agent/zmq_forwarder.rb', line 42

def self.context
  @@context_mutex.synchronize do
    @@zmq_context ||=
      begin
        require 'ffi-rzmq'
        context = ZMQ::Context.new(1)
        pid = Process.pid
        at_exit { context.terminate if Process.pid == pid }
        context
      end
  end
end

Instance Method Details

#connection_specs ⇒ Object 



22
23
24
25
26
 
# File 'lib/logjam_agent/zmq_forwarder.rb', line 22

def connection_specs
  @connection_specs ||= @config[:host].split(',').map do |host|
    augment_connection_spec(host, @config[:port])
  end
end

#default_options ⇒ Object 



28
29
30
31
32
33
34
35
36
37
 
# File 'lib/logjam_agent/zmq_forwarder.rb', line 28

def default_options
  {
    :port       => 9604,
    :linger     => 1000,
    :snd_hwm    => 1000,
    :rcv_hwm    => 1000,
    :rcv_timeo  => 5000,
    :snd_timeo  => 5000
  }
end

#ensure_ping_at_exit ⇒ Object 



61
62
63
64
65
66
 
# File 'lib/logjam_agent/zmq_forwarder.rb', line 61

def ensure_ping_at_exit
  return if @ping_ensured
  pid = Process.pid
  at_exit { (ping; reset) if Process.pid == pid }
  @ping_ensured = true
end

#forward(data, options = {}) ⇒ Object 



68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
 
# File 'lib/logjam_agent/zmq_forwarder.rb', line 68

def forward(data, options={})
  app_env = options[:app_env] || @app_env
  key = options[:routing_key] || "logs.#{app_env.sub('-','.')}"
  if engine = options[:engine]
    key += ".#{engine}"
  end
  msg = LogjamAgent.encode_payload(data)
  @socket_mutex.synchronize do
    if options[:sync]
      send_receive(app_env, key, msg)
    else
      publish(app_env, key, msg)
    end
  end
rescue => error
  reraise_expectation_errors!
  raise ForwardingError.new(error.message)
end

#reset ⇒ Object 



55
56
57
58
59
 
# File 'lib/logjam_agent/zmq_forwarder.rb', line 55

def reset
  @socket_mutex.synchronize do
    reset_without_locking
  end
end