Class: Riemann::Client::TCP

Inherits:
Riemann::Client show all
Defined in:
lib/riemann/client/tcp.rb

Constant Summary

Constants inherited from Riemann::Client

HOST, PORT, TIMEOUT

Instance Attribute Summary collapse

Attributes inherited from Riemann::Client

#tcp, #udp

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from Riemann::Client

#<<, #[], #connect, #query, #timeout

Constructor Details

#initialize(options = {}) ⇒ TCP

Returns a new instance of TCP.



20
21
22
23
 
# File 'lib/riemann/client/tcp.rb', line 20

def initialize(options = {})
  @options = options
  @locket  = Monitor.new
end

Instance Attribute Details

#host ⇒ Object

Returns the value of attribute host.



7
8
9
 
# File 'lib/riemann/client/tcp.rb', line 7

def host
  @host
end

#port ⇒ Object

Returns the value of attribute port.



7
8
9
 
# File 'lib/riemann/client/tcp.rb', line 7

def port
  @port
end

#socket ⇒ Object

Returns the value of attribute socket.



7
8
9
 
# File 'lib/riemann/client/tcp.rb', line 7

def socket
  @socket
end

Class Method Details

.socket_factory ⇒ Object

Public: Return a socket factory



16
17
18
 
# File 'lib/riemann/client/tcp.rb', line 16

def self.socket_factory
  @socket_factory || proc { |options| TcpSocket.connect(options) }
end

.socket_factory=(factory) ⇒ Object

Public: Set a socket factory – an object responding to #call(options) that returns a Socket object



11
12
13
 
# File 'lib/riemann/client/tcp.rb', line 11

def self.socket_factory=(factory)
  @socket_factory = factory
end

Instance Method Details

#close ⇒ Object 



40
41
42
43
44
45
 
# File 'lib/riemann/client/tcp.rb', line 40

def close
  @locket.synchronize do
    @socket.close if connected?
    @socket = nil
  end
end

#connected? ⇒ Boolean

Returns:

  • (Boolean) 


47
48
49
50
51
 
# File 'lib/riemann/client/tcp.rb', line 47

def connected?
  @locket.synchronize do
    !@socket.nil? && !@socket.closed?
  end
end

#read_message(s) ⇒ Object

Read a message from a stream



54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/riemann/client/tcp.rb', line 54

def read_message(s)
  if buffer = s.read(4) and buffer.size == 4
    length = buffer.unpack('N').first
    begin
      str = s.read length
      message = Riemann::Message.decode str
    rescue => e
      puts "Message was #{str.inspect}"
      raise
    end

    unless message.ok
      puts "Failed"
      raise ServerError, message.error
    end

    message
  else
    raise InvalidResponse, "unexpected EOF"
  end
end

#send_recv(message) ⇒ Object Also known as: send_maybe_recv 



76
77
78
79
80
81
 
# File 'lib/riemann/client/tcp.rb', line 76

def send_recv(message)
  with_connection do |s|
    s.write(message.encode_with_length)
    read_message(s)
  end
end

#with_connection ⇒ Object

Yields a connection in the block.



86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
 
# File 'lib/riemann/client/tcp.rb', line 86

def with_connection
  tries = 0

  @locket.synchronize do
    begin
      tries += 1
      yield(socket)
    rescue IOError, Errno::EPIPE, Errno::ECONNREFUSED, InvalidResponse, Timeout::Error, Riemann::Client::TcpSocket::Error
      close
      raise if tries > 3
      retry
    rescue Exception
      close
      raise
    end
  end
end