Class: Riemann::Client::TcpSocket

Inherits:
Object
  • Object
show all
Defined in:
lib/riemann/client/tcp_socket.rb

Overview

Socket: A specialized socket that has been configure

Defined Under Namespace

Classes: Error, Timeout

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(options = {}) ⇒ TcpSocket

Internal: Creates a new KJess::Socket



84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
 
# File 'lib/riemann/client/tcp_socket.rb', line 84

def initialize( options = {} )
  @host = options[:host]
  @port = options[:port]

  @connect_timeout = options[:connect_timeout] || options[:timeout] || 2
  @read_timeout    = options[:read_timeout]    || options[:timeout] || 2
  @write_timeout   = options[:write_timeout]   || options[:timeout] || 2

  @keepalive_active   = options.fetch(:keepalive_active, true)
  @keepalive_idle     = options[:keepalive_idle]     || 60
  @keepalive_interval = options[:keepalive_interval] || 30
  @keepalive_count    = options[:keepalive_count]    || 5

  @socket             = nil
end

Instance Attribute Details

#connect_timeout ⇒ Object (readonly)

Internal: The timeout for connecting in seconds. Defaults to 2



17
18
19
 
# File 'lib/riemann/client/tcp_socket.rb', line 17

def connect_timeout
  @connect_timeout
end

#host ⇒ Object (readonly)

Internal: The host this socket is connected to



25
26
27
 
# File 'lib/riemann/client/tcp_socket.rb', line 25

def host
  @host
end

#keepalive_count ⇒ Object (readonly)

Internal

Used for setting TCP_KEEPCNT: overrides tcp_keepalive_probes for a single socket.

tldp.org/HOWTO/TCP-Keepalive-HOWTO/usingkeepalive.html

tcp_keepalive_probes:

The number of unacknowledged probes to send before considering the
connection dead and notifying the application layer.



69
70
71
 
# File 'lib/riemann/client/tcp_socket.rb', line 69

def keepalive_count
  @keepalive_count
end

#keepalive_idle ⇒ Object (readonly)

Internal

Used for setting TCP_KEEPIDLE: overrides tcp_keepalive_time for a single socket.

tldp.org/HOWTO/TCP-Keepalive-HOWTO/usingkeepalive.html

tcp_keepalive_time:

The interval between the last data packet sent (simple ACKs are not
considered data) and the first keepalive probe; after the connection is
marked to need keepalive, this counter is not used any further.



43
44
45
 
# File 'lib/riemann/client/tcp_socket.rb', line 43

def keepalive_idle
  @keepalive_idle
end

#keepalive_interval ⇒ Object (readonly)

Internal

Used for setting TCP_KEEPINTVL: overrides tcp_keepalive_intvl for a single socket.

tldp.org/HOWTO/TCP-Keepalive-HOWTO/usingkeepalive.html

tcp_keepalive_intvl:

The interval between subsequential keepalive probes, regardless of what
the connection has exchanged in the meantime.



56
57
58
 
# File 'lib/riemann/client/tcp_socket.rb', line 56

def keepalive_interval
  @keepalive_interval
end

#port ⇒ Object (readonly)

Internal: The port this socket is connected to



29
30
31
 
# File 'lib/riemann/client/tcp_socket.rb', line 29

def port
  @port
end

#read_timeout ⇒ Object

Internal: The timeout for reading in seconds. Defaults to 2



13
14
15
 
# File 'lib/riemann/client/tcp_socket.rb', line 13

def read_timeout
  @read_timeout
end

#write_timeout ⇒ Object (readonly)

Internal: The timeout for writing in seconds. Defaults to 2



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

def write_timeout
  @write_timeout
end

Class Method Details

.connect(options = {}) ⇒ Object

Internal: Create and connect to the given location.

options, same as Constructor

Returns an instance of KJess::Socket



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

def self.connect(options = {})
  s = new(options)
  s.connect
  return s
end

Instance Method Details

#blank_socket ⇒ Object

Internal: Low level socket allocation and option configuration

Using the options from the initializer, a new ::Socket is created that is:

TCP, IPv4 only, autoclosing on exit, nagle's algorithm is disabled and has
TCP Keepalive options set if keepalive is supported.

Returns a new ::Socket instance



114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
 
# File 'lib/riemann/client/tcp_socket.rb', line 114

def blank_socket
  sock = ::Socket.new(::Socket::AF_INET, ::Socket::SOCK_STREAM, 0)

  # close file descriptors if we exec
  sock.fcntl(Fcntl::F_SETFD, Fcntl::FD_CLOEXEC)

  # Disable Nagle's algorithm
  sock.setsockopt(::Socket::IPPROTO_TCP, ::Socket::TCP_NODELAY, 1)

  if using_keepalive? then
    sock.setsockopt(::Socket::SOL_SOCKET, ::Socket::SO_KEEPALIVE , true)
    sock.setsockopt(::Socket::SOL_TCP,    ::Socket::TCP_KEEPIDLE , keepalive_idle)
    sock.setsockopt(::Socket::SOL_TCP,    ::Socket::TCP_KEEPINTVL, keepalive_interval)
    sock.setsockopt(::Socket::SOL_TCP,    ::Socket::TCP_KEEPCNT  , keepalive_count)
  end

  return sock
end

#close ⇒ Object

Internal: Closes the internal ::Socket

Returns nothing



146
147
148
149
 
# File 'lib/riemann/client/tcp_socket.rb', line 146

def close
  @socket.close unless closed?
  @socket = nil
end

#closed? ⇒ Boolean

Internal: Return true the socket is closed.

Returns:

  • (Boolean) 


152
153
154
155
156
 
# File 'lib/riemann/client/tcp_socket.rb', line 152

def closed?
  return true if @socket.nil?
  return true if @socket.closed?
  return false
end

#connect ⇒ Object

Internal:

Connect to the remote host in a non-blocking fashion.

Raise Error if there is a failure connecting.

Return the ::Socket on success



165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
 
# File 'lib/riemann/client/tcp_socket.rb', line 165

def connect
  # Calculate our timeout deadline
  deadline = Time.now.to_f + connect_timeout

  # Lookup destination address, we only want    IPv4             , TCP
  addrs      = ::Socket.getaddrinfo(host, port, ::Socket::AF_INET, ::Socket::SOCK_STREAM )
  errors     = []
  conn_error = lambda { raise errors.first }
  sock       = nil

  addrs.find( conn_error ) do |addr|
    sock = connect_or_error( addr, deadline, errors )
  end
  return sock
end

#connect_nonblock(addr, timeout) ⇒ Object

Internal: Connect to the give address within the timeout.

Make an attempt to connect to a single address within the given timeout.

Return the ::Socket when it is connected, or raise an Error if no connection was possible.



210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
 
# File 'lib/riemann/client/tcp_socket.rb', line 210

def connect_nonblock( addr, timeout )
  sockaddr = ::Socket.pack_sockaddr_in(addr[1], addr[3])
  sock     = blank_socket()
  sock.connect_nonblock( sockaddr )
  return sock
rescue Errno::EINPROGRESS
  if IO.select(nil, [sock], nil, timeout).nil?
    sock.close rescue nil
    raise Timeout, "Could not connect to #{host}:#{port} within #{timeout} seconds"
  end
  return connect_nonblock_finalize( sock, sockaddr )
rescue => ex
  sock.close rescue nil
  raise Error, "Could not connect to #{host}:#{port}: #{ex.class}: #{ex.message}", ex.backtrace
end

#connect_nonblock_finalize(sock, sockaddr) ⇒ Object

Internal: Make sure that a non-blocking connect has truely connected.

Ensure that the given socket is actually connected to the given adddress.

Returning the socket if it is and raising an Error if it isn’t.



232
233
234
235
236
237
238
239
240
 
# File 'lib/riemann/client/tcp_socket.rb', line 232

def connect_nonblock_finalize( sock, sockaddr )
  sock.connect_nonblock( sockaddr )
  return sock
rescue Errno::EISCONN
  return sock
rescue => ex
  sock.close rescue nil
  raise Error, "Could not connect to #{host}:#{port}: #{ex.class}: #{ex.message}", ex.backtrace
end

#connect_or_error(addr, deadline, errors) ⇒ Object

Internal: Connect to the destination or raise an error.

Connect to the address or capture the error of the connection

addr - An address returned from Socket.getaddrinfo() deadline - the after which we should raise a timeout error errors - a collection of errors to append an error too should we have one.

Make an attempt to connect to the given address. If it is successful, return the socket.

Should the connection fail, append the exception to the errors array and return false.



195
196
197
198
199
200
201
202
 
# File 'lib/riemann/client/tcp_socket.rb', line 195

def connect_or_error( addr, deadline, errors )
  timeout = deadline - Time.now.to_f
  raise Timeout, "Could not connect to #{host}:#{port}" if timeout <= 0
  return connect_nonblock( addr, timeout )
rescue Error => e
  errors << e
  return false
end

#keepalive_active? ⇒ Boolean

Internal: Return whether or not the keepalive_active flag is set.

Returns:

  • (Boolean) 


101
102
103
 
# File 'lib/riemann/client/tcp_socket.rb', line 101

def keepalive_active?
  @keepalive_active
end

#read(length, outbuf = nil) ⇒ Object

Reads length bytes from the socket

length - the number of bytes to read from the socket outbuf - an optional buffer to store the bytes in

Returns the bytes read if no outbuf is specified



269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
 
# File 'lib/riemann/client/tcp_socket.rb', line 269

def read(length, outbuf = nil)
  if outbuf
    outbuf.replace('')
    buf = outbuf
  else
    buf = ''
  end

  while buf.length < length
    unless rb = readpartial(length - buf.length)
      break
    end

    buf << rb
  end

  return buf
end

#readpartial(maxlen, outbuf = nil) ⇒ Object

Internal: Read up to a maxlen of data from the socket and store it in outbuf

maxlen - the maximum number of bytes to read from the socket outbuf - the buffer in which to store the bytes.

Returns the bytes read



294
295
296
297
298
299
300
301
302
 
# File 'lib/riemann/client/tcp_socket.rb', line 294

def readpartial(maxlen, outbuf = nil)
  return socket.read_nonblock(maxlen, outbuf)
rescue Errno::EWOULDBLOCK, Errno::EAGAIN, Errno::ECONNRESET
  if wait_readable(read_timeout)
    retry
  else
    raise Timeout, "Could not read from #{host}:#{port} in #{read_timeout} seconds"
  end
end

#socket ⇒ Object

Internal: Return the connected raw Socket.

If the socket is closed or non-existent it will create and connect again.

Returns a ::Socket



138
139
140
141
 
# File 'lib/riemann/client/tcp_socket.rb', line 138

def socket
  return @socket unless closed?
  @socket ||= connect()
end

#using_keepalive? ⇒ Boolean

Internal: say if we are using TCP Keep Alive or not

We will return true if the initialization options :keepalive_active is set to true, and if all the constants that are necessary to use TCP keep alive are defined.

It may be the case that on some operating systems that the constants are not defined, so in that case we do not want to attempt to use tcp keep alive if we are unable to do so in any case.

Returns true or false

Returns:

  • (Boolean) 


253
254
255
256
257
258
259
260
261
 
# File 'lib/riemann/client/tcp_socket.rb', line 253

def using_keepalive?
  using = false
  if keepalive_active? then
    using = [ :SOL_SOCKET, :SO_KEEPALIVE, :SOL_TCP, :TCP_KEEPIDLE, :TCP_KEEPINTVL, :TCP_KEEPCNT].all? do |c|
      ::Socket.const_defined? c
    end
  end
  return using
end

#wait_readable(timeout = nil) ⇒ Object 



329
330
331
 
# File 'lib/riemann/client/tcp_socket.rb', line 329

def wait_readable(timeout = nil)
  IO.select([@socket], nil, nil, timeout || read_timeout)
end

#wait_writable(timeout = nil) ⇒ Object 



325
326
327
 
# File 'lib/riemann/client/tcp_socket.rb', line 325

def wait_writable(timeout = nil)
  IO.select(nil, [@socket], nil, timeout || write_timeout)
end

#write(buf) ⇒ Object

Internal: Write the given data to the socket

buf - the data to write to the socket.

Raises an error if it is unable to write the data to the socket within the write_timeout.

returns nothing



312
313
314
315
316
317
318
319
320
321
322
323
 
# File 'lib/riemann/client/tcp_socket.rb', line 312

def write(buf)
  until buf.nil? or (buf.length == 0) do
    written = socket.write_nonblock(buf)
    buf = buf[written, buf.length]
  end
rescue Errno::EWOULDBLOCK, Errno::EINTR, Errno::EAGAIN, Errno::ECONNRESET
  if wait_writable(write_timeout)
    retry
  else
    raise Timeout, "Could not write to #{host}:#{port} in #{write_timeout} seconds"
  end
end