Class: Dalli::Protocol::Binary::ResponseProcessor
- Inherits:
-
Object
- Object
- Dalli::Protocol::Binary::ResponseProcessor
- Defined in:
- lib/dalli/protocol/binary/response_processor.rb
Overview
Class that encapsulates logic for processing binary protocol responses from memcached. Includes logic for pulling data from an IO source and parsing into local values. Handles errors on unexpected values.
Constant Summary collapse
- RESPONSE_CODES =
Response codes taken from: github.com/memcached/memcached/wiki/BinaryProtocolRevamped#response-status
{ 0 => 'No error', 1 => 'Key not found', 2 => 'Key exists', 3 => 'Value too large', 4 => 'Invalid arguments', 5 => 'Item not stored', 6 => 'Incr/decr on a non-numeric value', 7 => 'The vbucket belongs to another server', 8 => 'Authentication error', 9 => 'Authentication continue', 0x20 => 'Authentication required', 0x81 => 'Unknown command', 0x82 => 'Out of memory', 0x83 => 'Not supported', 0x84 => 'Internal error', 0x85 => 'Busy', 0x86 => 'Temporary failure' }.freeze
Instance Method Summary collapse
- #auth_response(buf = read_header) ⇒ Object
- #consume_all_responses_until_noop ⇒ Object
- #contains_header?(buf) ⇒ Boolean
- #data_cas_response ⇒ Object
-
#decr_incr ⇒ Object
Returns the new value for the key, if found and updated.
- #delete ⇒ Object
- #flush ⇒ Object
- #generic_response ⇒ Object
- #get(cache_nils: false) ⇒ Object
-
#getk_response_from_buffer(buf) ⇒ Object
This method returns an array of values used in a pipelined getk process.
-
#initialize(io_source, value_marshaller) ⇒ ResponseProcessor
constructor
A new instance of ResponseProcessor.
- #no_body_response ⇒ Object
- #raise_on_not_ok!(resp_header) ⇒ Object
- #read(num_bytes) ⇒ Object
- #read_header ⇒ Object
- #read_response ⇒ Object
- #reset ⇒ Object
- #response_header_from_buffer(buf) ⇒ Object
- #stats ⇒ Object
-
#storage_response ⇒ Object
Response for a storage operation.
- #unpack_response_body(resp_header, body, parse_as_stored_value) ⇒ Object
- #validate_auth_format(extra_len, count) ⇒ Object
- #version ⇒ Object
Constructor Details
#initialize(io_source, value_marshaller) ⇒ ResponseProcessor
Returns a new instance of ResponseProcessor.
34 35 36 37 |
# File 'lib/dalli/protocol/binary/response_processor.rb', line 34 def initialize(io_source, value_marshaller) @io_source = io_source @value_marshaller = value_marshaller end |
Instance Method Details
#auth_response(buf = read_header) ⇒ Object
184 185 186 187 188 189 190 |
# File 'lib/dalli/protocol/binary/response_processor.rb', line 184 def auth_response(buf = read_header) resp_header = ResponseHeader.new(buf) body_len = resp_header.body_len validate_auth_format(resp_header.extra_len, body_len) content = read(body_len) if body_len.positive? [resp_header.status, content] end |
#consume_all_responses_until_noop ⇒ Object
150 151 152 153 154 155 156 |
# File 'lib/dalli/protocol/binary/response_processor.rb', line 150 def consume_all_responses_until_noop loop do resp_header, = read_response # This is the response to the terminating noop / end of stat return true if resp_header.ok? && resp_header.key_len.zero? end end |
#contains_header?(buf) ⇒ Boolean
192 193 194 195 196 |
# File 'lib/dalli/protocol/binary/response_processor.rb', line 192 def contains_header?(buf) return false unless buf buf.bytesize >= ResponseHeader::SIZE end |
#data_cas_response ⇒ Object
103 104 105 106 107 108 109 110 111 112 |
# File 'lib/dalli/protocol/binary/response_processor.rb', line 103 def data_cas_response resp_header, body = read_response return [nil, resp_header.cas] if resp_header.not_found? return [nil, false] if resp_header.not_stored? raise_on_not_ok!(resp_header) return [nil, resp_header.cas] unless body [unpack_response_body(resp_header, body, true).last, resp_header.cas] end |
#decr_incr ⇒ Object
Returns the new value for the key, if found and updated
115 116 117 118 |
# File 'lib/dalli/protocol/binary/response_processor.rb', line 115 def decr_incr body = generic_response body ? body.unpack1('Q>') : body end |
#delete ⇒ Object
95 96 97 98 99 100 101 |
# File 'lib/dalli/protocol/binary/response_processor.rb', line 95 def delete resp_header, = read_response return false if resp_header.not_found? || resp_header.not_stored? raise_on_not_ok!(resp_header) true end |
#flush ⇒ Object
138 139 140 |
# File 'lib/dalli/protocol/binary/response_processor.rb', line 138 def flush no_body_response end |
#generic_response ⇒ Object
158 159 160 161 162 163 164 165 166 167 168 |
# File 'lib/dalli/protocol/binary/response_processor.rb', line 158 def generic_response resp_header, body = read_response return false if resp_header.not_stored? # Not stored, normal status for add operation return nil if resp_header.not_found? raise_on_not_ok!(resp_header) return true unless body unpack_response_body(resp_header, body, false).last end |
#get(cache_nils: false) ⇒ Object
69 70 71 72 73 74 75 76 77 78 79 |
# File 'lib/dalli/protocol/binary/response_processor.rb', line 69 def get(cache_nils: false) resp_header, body = read_response return false if resp_header.not_stored? # Not stored, normal status for add operation return cache_nils ? ::Dalli::NOT_FOUND : nil if resp_header.not_found? raise_on_not_ok!(resp_header) return true unless body unpack_response_body(resp_header, body, true).last end |
#getk_response_from_buffer(buf) ⇒ Object
This method returns an array of values used in a pipelined getk process. The first value is the number of bytes by which to advance the pointer in the buffer. If the complete response is found in the buffer, this will be the response size. Otherwise it is zero.
The remaining three values in the array are the ResponseHeader, key, and value.
212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 |
# File 'lib/dalli/protocol/binary/response_processor.rb', line 212 def getk_response_from_buffer(buf) # There's no header in the buffer, so don't advance return [0, nil, nil, nil, nil] unless contains_header?(buf) resp_header = response_header_from_buffer(buf) body_len = resp_header.body_len # We have a complete response that has no body. # This is either the response to the terminating # noop or, if the status is not zero, an intermediate # error response that needs to be discarded. return [ResponseHeader::SIZE, resp_header.ok?, resp_header.cas, nil, nil] if body_len.zero? resp_size = ResponseHeader::SIZE + body_len # The header is in the buffer, but the body is not. As we don't have # a complete response, don't advance the buffer return [0, nil, nil, nil, nil] unless buf.bytesize >= resp_size # The full response is in our buffer, so parse it and return # the values body = buf.byteslice(ResponseHeader::SIZE, body_len) key, value = unpack_response_body(resp_header, body, true) [resp_size, resp_header.ok?, resp_header.cas, key, value] end |
#no_body_response ⇒ Object
170 171 172 173 174 175 176 |
# File 'lib/dalli/protocol/binary/response_processor.rb', line 170 def no_body_response resp_header, = read_response return false if resp_header.not_stored? # Not stored, possible status for append/prepend/delete raise_on_not_ok!(resp_header) true end |
#raise_on_not_ok!(resp_header) ⇒ Object
63 64 65 66 67 |
# File 'lib/dalli/protocol/binary/response_processor.rb', line 63 def raise_on_not_ok!(resp_header) return if resp_header.ok? raise Dalli::DalliError, "Response error #{resp_header.status}: #{RESPONSE_CODES[resp_header.status]}" end |
#read(num_bytes) ⇒ Object
39 40 41 |
# File 'lib/dalli/protocol/binary/response_processor.rb', line 39 def read(num_bytes) @io_source.read(num_bytes) end |
#read_header ⇒ Object
59 60 61 |
# File 'lib/dalli/protocol/binary/response_processor.rb', line 59 def read_header read(ResponseHeader::SIZE) || raise(Dalli::NetworkError, 'No response') end |
#read_response ⇒ Object
43 44 45 46 47 |
# File 'lib/dalli/protocol/binary/response_processor.rb', line 43 def read_response resp_header = ResponseHeader.new(read_header) body = read(resp_header.body_len) if resp_header.body_len.positive? [resp_header, body] end |
#reset ⇒ Object
142 143 144 |
# File 'lib/dalli/protocol/binary/response_processor.rb', line 142 def reset generic_response end |
#response_header_from_buffer(buf) ⇒ Object
198 199 200 |
# File 'lib/dalli/protocol/binary/response_processor.rb', line 198 def response_header_from_buffer(buf) ResponseHeader.new(buf) end |
#stats ⇒ Object
120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 |
# File 'lib/dalli/protocol/binary/response_processor.rb', line 120 def stats hash = {} loop do resp_header, body = read_response # This is the response to the terminating noop / end of stat return hash if resp_header.ok? && resp_header.key_len.zero? # Ignore any responses with non-zero status codes, # such as errors from set operations. That allows # this code to be used at the end of a multi # block to clear any error responses from inside the multi. next unless resp_header.ok? key, value = unpack_response_body(resp_header, body, true) hash[key] = value end end |
#storage_response ⇒ Object
Response for a storage operation. Returns the cas on success. False if the value wasn’t stored. And raises an error on all other error codes from memcached.
86 87 88 89 90 91 92 93 |
# File 'lib/dalli/protocol/binary/response_processor.rb', line 86 def storage_response resp_header, = read_response return nil if resp_header.not_found? return false if resp_header.not_stored? # Not stored, normal status for add operation raise_on_not_ok!(resp_header) resp_header.cas end |
#unpack_response_body(resp_header, body, parse_as_stored_value) ⇒ Object
49 50 51 52 53 54 55 56 57 |
# File 'lib/dalli/protocol/binary/response_processor.rb', line 49 def unpack_response_body(resp_header, body, parse_as_stored_value) extra_len = resp_header.extra_len key_len = resp_header.key_len bitflags = extra_len.positive? ? body.unpack1('N') : 0x0 key = body.byteslice(extra_len, key_len).force_encoding(Encoding::UTF_8) if key_len.positive? value = body.byteslice((extra_len + key_len)..-1) value = @value_marshaller.retrieve(value, bitflags) if parse_as_stored_value [key, value] end |
#validate_auth_format(extra_len, count) ⇒ Object
178 179 180 181 182 |
# File 'lib/dalli/protocol/binary/response_processor.rb', line 178 def validate_auth_format(extra_len, count) return if extra_len.zero? raise Dalli::NetworkError, "Unexpected message format: #{extra_len} #{count}" end |
#version ⇒ Object
146 147 148 |
# File 'lib/dalli/protocol/binary/response_processor.rb', line 146 def version generic_response end |