Class: KJess::Socket

Inherits:
Object
  • Object
show all
Defined in:
lib/kjess/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 = {}) ⇒ Socket

Internal: Creates a new KJess::Socket



82
83
84
85
86
87
88
89
90
91
92
93
 
# File 'lib/kjess/socket.rb', line 82

def initialize( options = {} )
  @host               = options.fetch(:host)
  @port               = options.fetch(:port)
  @connect_timeout    = options.fetch(:connect_timeout   ,  2)
  @read_timeout       = options.fetch(:read_timeout      ,  2)
  @write_timeout      = options.fetch(:write_timeout     ,  2)
  @keepalive_active   = options.fetch(:keepalive_active  , true)
  @keepalive_idle     = options.fetch(:keepalive_idle    , 60)
  @keepalive_interval = options.fetch(:keepalive_interval, 30)
  @keepalive_count    = options.fetch(:keepalive_count   ,  5)
  @socket             = nil
end

Instance Attribute Details

#connect_timeout ⇒ Object (readonly)

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



15
16
17
 
# File 'lib/kjess/socket.rb', line 15

def connect_timeout
  @connect_timeout
end

#host ⇒ Object (readonly)

Internal: The host this socket is connected to



23
24
25
 
# File 'lib/kjess/socket.rb', line 23

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.



67
68
69
 
# File 'lib/kjess/socket.rb', line 67

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.



41
42
43
 
# File 'lib/kjess/socket.rb', line 41

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.



54
55
56
 
# File 'lib/kjess/socket.rb', line 54

def keepalive_interval
  @keepalive_interval
end

#port ⇒ Object (readonly)

Internal: The port this socket is connected to



27
28
29
 
# File 'lib/kjess/socket.rb', line 27

def port
  @port
end

#read_timeout ⇒ Object

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



11
12
13
 
# File 'lib/kjess/socket.rb', line 11

def read_timeout
  @read_timeout
end

#write_timeout ⇒ Object (readonly)

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



19
20
21
 
# File 'lib/kjess/socket.rb', line 19

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



75
76
77
78
79
 
# File 'lib/kjess/socket.rb', line 75

def self.connect( options = {} )
  s = Socket.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



109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
 
# File 'lib/kjess/socket.rb', line 109

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



141
142
143
144
 
# File 'lib/kjess/socket.rb', line 141

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

#closed? ⇒ Boolean

Internal: Return true the socket is closed.

Returns:

  • (Boolean) 


147
148
149
150
151
 
# File 'lib/kjess/socket.rb', line 147

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



160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
 
# File 'lib/kjess/socket.rb', line 160

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.



205
206
207
208
209
210
211
212
213
214
215
216
217
 
# File 'lib/kjess/socket.rb', line 205

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 !wait_writable(timeout, sock)
    raise Timeout, "Could not connect to #{host}:#{port} within #{timeout} seconds"
  end
  return connect_nonblock_finalize( sock, sockaddr )
rescue => ex
  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.



225
226
227
228
229
230
231
232
 
# File 'lib/kjess/socket.rb', line 225

def connect_nonblock_finalize( sock, sockaddr )
  sock.connect_nonblock( sockaddr )
  return sock
rescue Errno::EISCONN
  return sock
rescue => ex
  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.



190
191
192
193
194
195
196
197
 
# File 'lib/kjess/socket.rb', line 190

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) 


96
97
98
 
# File 'lib/kjess/socket.rb', line 96

def keepalive_active?
  @keepalive_active
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



261
262
263
264
265
266
267
268
269
 
# File 'lib/kjess/socket.rb', line 261

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



133
134
135
136
 
# File 'lib/kjess/socket.rb', line 133

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) 


245
246
247
248
249
250
251
252
253
 
# File 'lib/kjess/socket.rb', line 245

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, socket = @socket) ⇒ Object 



296
297
298
 
# File 'lib/kjess/socket.rb', line 296

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

#wait_writable(timeout = nil, socket = @socket) ⇒ Object 



292
293
294
 
# File 'lib/kjess/socket.rb', line 292

def wait_writable(timeout = nil, socket = @socket)
  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



279
280
281
282
283
284
285
286
287
288
289
290
 
# File 'lib/kjess/socket.rb', line 279

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