Class: LogjamAgent::ZMQForwarder
- Inherits:
-
Object
- Object
- LogjamAgent::ZMQForwarder
show all
- Includes:
- Util
- Defined in:
- lib/logjam_agent/zmq_forwarder.rb
Constant Summary
collapse
- SEQUENCE_START =
0
- @@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
Returns a new instance of ZMQForwarder.
9
10
11
12
13
14
15
16
17
|
# File 'lib/logjam_agent/zmq_forwarder.rb', line 9
def initialize(*args)
opts = args.
@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
end
|
Instance Attribute Details
#app ⇒ Object
Returns the value of attribute app.
3
4
5
|
# File 'lib/logjam_agent/zmq_forwarder.rb', line 3
def app
@app
end
|
#env ⇒ Object
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
39
40
41
42
43
44
45
46
47
48
49
|
# File 'lib/logjam_agent/zmq_forwarder.rb', line 39
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
#connection_specs ⇒ Object
19
20
21
22
23
|
# File 'lib/logjam_agent/zmq_forwarder.rb', line 19
def connection_specs
@connection_specs ||= @config[:host].split(',').map do |host|
augment_connection_spec(host, @config[:port])
end
end
|
#default_options ⇒ Object
25
26
27
28
29
30
31
32
33
34
|
# File 'lib/logjam_agent/zmq_forwarder.rb', line 25
def default_options
{
:port => 9604,
:linger => 1000,
:snd_hwm => 100,
:rcv_hwm => 100,
:rcv_timeo => 5000,
:snd_timeo => 5000
}
end
|
#forward(data, options = {}) ⇒ Object
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
|
# File 'lib/logjam_agent/zmq_forwarder.rb', line 72
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)
if options[:sync]
send_receive(app_env, key, msg)
else
publish(app_env, key, msg)
end
rescue => error
reraise_expectation_errors!
raise ForwardingError.new(error.message)
end
|
#publish(app_env, key, data) ⇒ Object
89
90
91
92
93
94
95
|
# File 'lib/logjam_agent/zmq_forwarder.rb', line 89
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 on publishing: #{ZMQ::Util.error_string}"
end
end
|
#reset ⇒ Object
65
66
67
68
69
70
|
# File 'lib/logjam_agent/zmq_forwarder.rb', line 65
def reset
if @socket
@socket.close
@socket = nil
end
end
|
#socket ⇒ Object
51
52
53
54
55
56
57
58
59
60
61
62
63
|
# File 'lib/logjam_agent/zmq_forwarder.rb', line 51
def socket
return @socket if @socket
@socket = self.class.context.socket(ZMQ::DEALER)
at_exit { ping; reset }
@socket.setsockopt(ZMQ::LINGER, @config[:linger])
@socket.setsockopt(ZMQ::SNDHWM, @config[:snd_hwm])
@socket.setsockopt(ZMQ::RCVHWM, @config[:rcv_hwm])
@socket.setsockopt(ZMQ::RCVTIMEO, @config[:rcv_timeo])
@socket.setsockopt(ZMQ::SNDTIMEO, @config[:snd_timeo])
spec = connection_specs.sort_by{rand}.first
@socket.connect(spec)
@socket
end
|