Class: RedisClient::RubyConnection::BufferedIO

Inherits:
Object
  • Object
show all
Defined in:
lib/redis_client/ruby_connection/buffered_io.rb

Constant Summary collapse

EOL = 
"\r\n".b.freeze
EOL_SIZE = 
EOL.bytesize
ENCODING = 
Encoding::BINARY

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(io, read_timeout:, write_timeout:, chunk_size: 4096) ⇒ BufferedIO

Returns a new instance of BufferedIO.



16
17
18
19
20
21
22
23
24
 
# File 'lib/redis_client/ruby_connection/buffered_io.rb', line 16

def initialize(io, read_timeout:, write_timeout:, chunk_size: 4096)
  @io = io
  @buffer = +""
  @offset = 0
  @chunk_size = chunk_size
  @read_timeout = read_timeout
  @write_timeout = write_timeout
  @blocking_reads = false
end

Instance Attribute Details

#read_timeout ⇒ Object

Returns the value of attribute read_timeout.



11
12
13
 
# File 'lib/redis_client/ruby_connection/buffered_io.rb', line 11

def read_timeout
  @read_timeout
end

#write_timeout ⇒ Object

Returns the value of attribute write_timeout.



11
12
13
 
# File 'lib/redis_client/ruby_connection/buffered_io.rb', line 11

def write_timeout
  @write_timeout
end

Instance Method Details

#close ⇒ Object 



89
90
91
 
# File 'lib/redis_client/ruby_connection/buffered_io.rb', line 89

def close
  @io.to_io.close
end

#closed? ⇒ Boolean

Returns:

  • (Boolean) 


93
94
95
 
# File 'lib/redis_client/ruby_connection/buffered_io.rb', line 93

def closed?
  @io.to_io.closed?
end

#eof? ⇒ Boolean

Returns:

  • (Boolean) 


97
98
99
 
# File 'lib/redis_client/ruby_connection/buffered_io.rb', line 97

def eof?
  @offset >= @buffer.bytesize && @io.eof?
end

#getbyte ⇒ Object 



150
151
152
153
154
155
156
157
 
# File 'lib/redis_client/ruby_connection/buffered_io.rb', line 150

def getbyte
  unless byte = @buffer.getbyte(@offset)
    ensure_remaining(1)
    byte = @buffer.getbyte(@offset)
  end
  @offset += 1
  byte
end

#gets_chomp ⇒ Object 



26
27
28
29
30
31
32
33
34
35
 
# File 'lib/redis_client/ruby_connection/buffered_io.rb', line 26

def gets_chomp
  fill_buffer(false) if @offset >= @buffer.bytesize
  until eol_index = @buffer.byteindex(EOL, @offset)
    fill_buffer(false)
  end

  line = @buffer.byteslice(@offset, eol_index - @offset)
  @offset = eol_index + EOL_SIZE
  line
end

#gets_integer ⇒ Object 



159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
 
# File 'lib/redis_client/ruby_connection/buffered_io.rb', line 159

def gets_integer
  int = 0
  offset = @offset
  while true
    chr = @buffer.getbyte(offset)

    if chr
      if chr == 13 # "\r".ord
        @offset = offset + 2
        break
      else
        int = (int * 10) + chr - 48
      end
      offset += 1
    else
      ensure_line
      return gets_integer
    end
  end

  int
end

#read_chomp(bytes) ⇒ Object 



37
38
39
40
41
42
 
# File 'lib/redis_client/ruby_connection/buffered_io.rb', line 37

def read_chomp(bytes)
  ensure_remaining(bytes + EOL_SIZE)
  str = @buffer.byteslice(@offset, bytes)
  @offset += bytes + EOL_SIZE
  str
end

#skip(offset) ⇒ Object 



121
122
123
124
125
 
# File 'lib/redis_client/ruby_connection/buffered_io.rb', line 121

def skip(offset)
  ensure_remaining(offset)
  @offset += offset
  nil
end

#with_timeout(new_timeout) ⇒ Object 



101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
 
# File 'lib/redis_client/ruby_connection/buffered_io.rb', line 101

def with_timeout(new_timeout)
  new_timeout = false if new_timeout == 0

  previous_read_timeout = @read_timeout
  previous_blocking_reads = @blocking_reads

  if new_timeout
    @read_timeout = new_timeout
  else
    @blocking_reads = true
  end

  begin
    yield
  ensure
    @read_timeout = previous_read_timeout
    @blocking_reads = previous_blocking_reads
  end
end

#write(string) ⇒ Object 



127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
 
# File 'lib/redis_client/ruby_connection/buffered_io.rb', line 127

def write(string)
  total = remaining = string.bytesize
  loop do
    case bytes_written = @io.write_nonblock(string, exception: false)
    when Integer
      remaining -= bytes_written
      if remaining > 0
        string = string.byteslice(bytes_written..-1)
      else
        return total
      end
    when :wait_readable
      @io.to_io.wait_readable(@read_timeout) or raise(ReadTimeoutError, "Waited #{@read_timeout} seconds")
    when :wait_writable
      @io.to_io.wait_writable(@write_timeout) or raise(WriteTimeoutError, "Waited #{@write_timeout} seconds")
    when nil
      raise Errno::ECONNRESET
    else
      raise "Unexpected `write_nonblock` return: #{bytes.inspect}"
    end
  end
end