Class: LogjamAgent::ZMQForwarder

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

Constant Summary collapse

@@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

#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.



7
8
9
10
11
12
13
14
15
16
 
# File 'lib/logjam_agent/zmq_forwarder.rb', line 7

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)
  @zmq_hosts = Array(@config[:host])
  @zmq_port = @config[:port]
  @sequence = 0
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 



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

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

Instance Method Details

#default_options ⇒ Object 



18
19
20
21
22
23
 
# File 'lib/logjam_agent/zmq_forwarder.rb', line 18

def default_options
  {
    :host         => "localhost",
    :port         => 12345
  }
end

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



60
61
62
63
64
65
66
67
68
69
70
71
72
73
 
# File 'lib/logjam_agent/zmq_forwarder.rb', line 60

def forward(msg, options={})
  return if LogjamAgent.disabled
  begin
    app_env = options[:app_env] || @app_env
    key = options[:routing_key] || "logs.#{app_env.sub('-','.')}"
    if engine = options[:engine]
      key += ".#{engine}"
    end
    publish(app_env, key, msg)
  rescue => error
    reraise_expectation_errors!
    raise ForwardingError.new(error.message)
  end
end

#publish(app_env, key, data) ⇒ Object 



75
76
77
78
79
80
81
 
# File 'lib/logjam_agent/zmq_forwarder.rb', line 75

def publish(app_env, key, data)
  info = pack_info(@sequence = next_fixnum(@sequence))
  parts = [app_env, key, data, info]
  if socket.send_strings(parts, ZMQ::DONTWAIT) < 0
    raise "ZMQ error: #{ZMQ::Util.error_string}"
  end
end

#reset ⇒ Object 



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

def reset
  return unless @socket
  @socket.close
  @socket = nil
end

#socket ⇒ Object 



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

def socket
  @socket ||=
    begin
      socket = self.class.context.socket(ZMQ::PUSH)
      socket.setsockopt(ZMQ::LINGER, 100)
      socket.setsockopt(ZMQ::SNDHWM, 10)
      @zmq_hosts.each do |host|
        socket.connect("tcp://#{host}:#{@zmq_port}")
      end
      at_exit { reset }
      socket
    end
end