Sha256: f5e3618093c6c68bf3f78c36287f22f732d696ce1c00c783e7bc40253b8627b3
Contents?: true
Size: 954 Bytes
Versions: 11
Compression:
Stored size: 954 Bytes
Contents
require "timeout"
require "socket"
require "kurchatov/riemann/message"
module Kurchatov
module Riemann
class Client
attr_accessor :host, :port
CONNECT_TIMEOUT = 5
SEND_TIMEOUT = 5
RIEMANN_PORT = 5555
def initialize(opts = {})
@host = opts[:host]
@port = opts[:port] || RIEMANN_PORT
@mutex = Mutex.new
end
def <<(events)
events = events.map {|e| Event.new(e) }
message = Message.new(:events => events)
with_connection do |socket|
x = message.encode_with_length
Timeout::timeout(SEND_TIMEOUT) {
socket.write(x)
socket.flush
}
end
end
def with_connection
@mutex.synchronize do
yield(@socket || connect)
end
end
def connect
Timeout::timeout(CONNECT_TIMEOUT) {
@socket ||= TCPSocket.new(@host, @port)
}
end
end
end
end
Version data entries
11 entries across 11 versions & 1 rubygems