Module: Bunny::Socket

Defined in:
lib/bunny/cruby/socket.rb

Overview

TCP socket extension that uses TCP_NODELAY and supports reading fully.

Heavily inspired by Dalli by Mike Perham.

Constant Summary collapse

READ_RETRY_EXCEPTION_CLASSES = 
[Errno::EAGAIN, Errno::EWOULDBLOCK, IO::WaitReadable,
IO::EAGAINWaitReadable, IO::EWOULDBLOCKWaitReadable].freeze
WRITE_RETRY_EXCEPTION_CLASSES = 
[Errno::EAGAIN, Errno::EWOULDBLOCK, IO::WaitWritable,
IO::EAGAINWaitWritable, IO::EWOULDBLOCKWaitWritable].freeze

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Instance Attribute Details

#options ⇒ Object

Returns the value of attribute options.



12
13
14
 
# File 'lib/bunny/cruby/socket.rb', line 12

def options
  @options
end

Class Method Details

.open(host, port, options = {}) ⇒ Object 



19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
 
# File 'lib/bunny/cruby/socket.rb', line 19

def self.open(host, port, options = {})
  socket = ::Socket.tcp(host, port, nil, nil,
                        connect_timeout: options[:connect_timeout],
                        resolv_timeout: options[:connect_timeout])
  if ::Socket.constants.include?('TCP_NODELAY') || ::Socket.constants.include?(:TCP_NODELAY)
    socket.setsockopt(::Socket::IPPROTO_TCP, ::Socket::TCP_NODELAY, true)
  end
  socket.setsockopt(::Socket::SOL_SOCKET, ::Socket::SO_KEEPALIVE, true) if options.fetch(:keepalive, true)
  socket.instance_eval do
    @__bunny_socket_eof_flag__ = false
  end
  socket.extend self
  socket.options = { host: host, port: port }.merge(options)
  socket
rescue Errno::ETIMEDOUT
  raise ClientTimeout
end

Instance Method Details

#read_fully(count, timeout = nil) ⇒ String

Reads given number of bytes with an optional timeout

Parameters:

  • count (Integer) —

    How many bytes to read

  • timeout (Integer) (defaults to: nil) —

    Timeout

Returns:

  • (String) —

    Data read from the socket



44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
 
# File 'lib/bunny/cruby/socket.rb', line 44

def read_fully(count, timeout = nil)
  return nil if @__bunny_socket_eof_flag__

  value = +''
  begin
    loop do
      value << read_nonblock(count - value.bytesize)
      break if value.bytesize >= count
    end
  rescue EOFError
    # @eof will break Rubinius' TCPSocket implementation. MK.
    @__bunny_socket_eof_flag__ = true
  rescue *READ_RETRY_EXCEPTION_CLASSES
    if IO.select([self], nil, nil, timeout)
      retry
    else
      raise Timeout::Error, "IO timeout when reading #{count} bytes"
    end
  end
  value
end

#read_fully_into(buffer, count, timeout = nil) ⇒ String

Reads given number of bytes into an existing buffer with an optional timeout

Parameters:

  • buffer (String) —

    Buffer to read into (will be appended to)

  • count (Integer) —

    How many bytes to read

  • timeout (Integer) (defaults to: nil) —

    Timeout

Returns:

  • (String) —

    The buffer with data appended



74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
 
# File 'lib/bunny/cruby/socket.rb', line 74

def read_fully_into(buffer, count, timeout = nil)
  return nil if @__bunny_socket_eof_flag__

  bytes_read = 0
  begin
    loop do
      chunk = read_nonblock(count - bytes_read)
      buffer << chunk
      bytes_read += chunk.bytesize
      break if bytes_read >= count
    end
  rescue EOFError
    @__bunny_socket_eof_flag__ = true
  rescue *READ_RETRY_EXCEPTION_CLASSES
    if IO.select([self], nil, nil, timeout)
      retry
    else
      raise Timeout::Error, "IO timeout when reading #{count} bytes"
    end
  end
  buffer
end

#write_nonblock_fully(data, timeout = nil) ⇒ Object

Writes provided data using IO#write_nonblock, taking care of handling of exceptions it raises when writing would fail (e.g. due to socket buffer being full).

IMPORTANT: this method will mutate (slice) the argument. Pass in duplicates if this is not appropriate in your case.

Parameters:

  • data (String) —

    Data to write

  • timeout (Integer) (defaults to: nil) —

    Timeout



108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
 
# File 'lib/bunny/cruby/socket.rb', line 108

def write_nonblock_fully(data, timeout = nil)
  return nil if @__bunny_socket_eof_flag__

  length = data.bytesize
  total_count = 0
  count = 0
  loop do
    begin
      count = self.write_nonblock(data)
    rescue *WRITE_RETRY_EXCEPTION_CLASSES
      if IO.select([], [self], nil, timeout)
        retry
      else
        raise Timeout::Error, "IO timeout when writing to socket"
      end
    end

    total_count += count
    return total_count if total_count >= length
    data = data.byteslice(count..-1)
  end

end