Class: KJess::Connection

Inherits:
Object
  • Object
show all
Defined in:
lib/kjess/connection.rb

Overview

Connection

Defined Under Namespace

Classes: Error

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(host, port = 22133, options = {}) ⇒ Connection

TODO: make port an option at next major version number change



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

def initialize( host, port = 22133, options = {} )
  if port.is_a?(Hash)
    options = port
    port = 22133
  end

  @options         = options.dup
  @options[:host] = host
  @options[:port] = Float( port ).to_i
  @socket          = nil
  @pid             = nil
  @read_buffer     = ''
end

Class Method Details

.default_socket_factory ⇒ Object

Internal: Returns the default socket factory



31
32
33
 
# File 'lib/kjess/connection.rb', line 31

def self.default_socket_factory
  lambda { |options| KJess::Socket.connect(options) }
end

.socket_factory ⇒ Object

Public: Return the socket factory



24
25
26
27
 
# File 'lib/kjess/connection.rb', line 24

def self.socket_factory
  @socket_factory ||= nil
  @socket_factory.respond_to?(:call) ? @socket_factory : default_socket_factory
end

.socket_factory=(factory) ⇒ Object

Public: Set a socket factory

factory - an object that responds to #call(options) where options is

a Hash.

returns nothing



18
19
20
 
# File 'lib/kjess/connection.rb', line 18

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

Instance Method Details

#close ⇒ Object

Internal: close the socket if it is not already closed

Returns nothing



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

def close
  @socket.close if @socket and not @socket.closed?
  @read_buffer = ''
  @socket = nil
end

#closed? ⇒ Boolean

Internal: is the socket closed

Returns true or false

Returns:

  • (Boolean) 


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

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

#connect_timeout ⇒ Object

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



46
47
48
 
# File 'lib/kjess/connection.rb', line 46

def connect_timeout
  socket.connect_timeout
end

#host ⇒ Object

Public: The hostname/ip address to connect to.



36
37
38
 
# File 'lib/kjess/connection.rb', line 36

def host
  @options[:host]
end

#keepalive_active? ⇒ Boolean

Internal: return thekeepalive timeout

Returns:

  • (Boolean) 


61
62
63
 
# File 'lib/kjess/connection.rb', line 61

def keepalive_active?
  socket.keepalive_active?
end

#keepalive_count ⇒ Object

Internal: return the keepalive count The keepalive count



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

def keepalive_count
  socket.keepalive_count
end

#keepalive_idle ⇒ Object

Internal: return the keepalive idle



77
78
79
 
# File 'lib/kjess/connection.rb', line 77

def keepalive_idle
  socket.keepalive_idle
end

#keepalive_interval ⇒ Object

Internal: return the keepalive interval



72
73
74
 
# File 'lib/kjess/connection.rb', line 72

def keepalive_interval
  socket.keepalive_interval
end

#port ⇒ Object

Public: The port number to connect to. Default 22133



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

def port
  @options[:port]
end

#read(nbytes) ⇒ Object

Internal: Read from the socket

nbytes - this method takes the number of bytes to read

Returns what IO#read returns



195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
 
# File 'lib/kjess/connection.rb', line 195

def read( nbytes )
  while @read_buffer.length < nbytes
    @read_buffer << socket.readpartial(nbytes - @read_buffer.length)
  end

  result = @read_buffer.slice!(0, nbytes)

  $stderr.puts "<-- #{result}" if $DEBUG
  return result
rescue KJess::NetworkError
  close
  raise
rescue => e
  close
  raise Error, "Could not read from #{host}:#{port}: #{e.class}: #{e.message}", e.backtrace
end

#read_timeout ⇒ Object

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



51
52
53
 
# File 'lib/kjess/connection.rb', line 51

def read_timeout
  socket.read_timeout
end

#readline(eom = Protocol::CRLF) ⇒ Object

Internal: read a single line from the socket

eom - the End Of Mesasge delimiter (default: “rn”)

Returns a String



168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
 
# File 'lib/kjess/connection.rb', line 168

def readline( eom = Protocol::CRLF )
  while true
    while (idx = @read_buffer.index(eom)) == nil
      @read_buffer << socket.readpartial(10240)
    end

    line = @read_buffer.slice!(0, idx + eom.length)
    $stderr.puts "<-- #{line}" if $DEBUG
    break unless line.strip.length == 0
  end
  return line
rescue KJess::NetworkError
  close
  raise
rescue EOFError
  close
  return "EOF"
rescue => e
  close
  raise Error, "Could not read from #{host}:#{port}: #{e.class}: #{e.message}", e.backtrace
end

#socket ⇒ Object

Internal: Return the socket that is connected to the Kestrel server

Returns the socket. If the socket is not connected it will connect and then return it.

Make sure that we close the socket if we are not the same process that opened that socket to begin with.

Returns a KJess::Socket



118
119
120
121
122
123
124
125
126
127
 
# File 'lib/kjess/connection.rb', line 118

def socket
  close if @pid && @pid != Process.pid
  return @socket if @socket and not @socket.closed?
  @socket      = self.class.socket_factory.call(@options)
  @pid         = Process.pid
  @read_buffer = ''
  return @socket
rescue => e
  raise Error, "Could not connect to #{host}:#{port}: #{e.class}: #{e.message}", e.backtrace
end

#with_additional_read_timeout(additional_timeout, &block) ⇒ Object

Internal: Adds time to the read timeout

additional_timeout - additional number of seconds to the read timeout

Returns nothing



101
102
103
104
105
106
107
 
# File 'lib/kjess/connection.rb', line 101

def with_additional_read_timeout(additional_timeout, &block)
  old_read_timeout = socket.read_timeout
  socket.read_timeout += additional_timeout
  block.call
ensure
  @read_timeout = old_read_timeout
end

#write(msg) ⇒ Object

Internal: write the given item to the socket

msg - the message to write

Returns nothing



152
153
154
155
156
157
158
159
160
161
 
# File 'lib/kjess/connection.rb', line 152

def write( msg )
  $stderr.puts "--> #{msg}" if $DEBUG
  socket.write( msg )
rescue KJess::NetworkError
  close
  raise
rescue => e
  close
  raise Error, "Could not write to #{host}:#{port}: #{e.class}: #{e.message}", e.backtrace
end

#write_timeout ⇒ Object

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



56
57
58
 
# File 'lib/kjess/connection.rb', line 56

def write_timeout
  socket.write_timeout
end