Method: Aerospike::ReadCommand#parse_result

Defined in:
lib/aerospike/command/read_command.rb

#parse_resultObject



52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
# File 'lib/aerospike/command/read_command.rb', line 52

def parse_result
  # Read header.
  begin
    @conn.read(@data_buffer, 8)
  rescue => e
    Aerospike.logger.error("parse result error: #{e}")
    raise e
  end

  # inflate if compressed
  compressed_sz = compressed_size
  if compressed_sz
    begin
      # waste 8 size bytes
      @conn.read(@data_buffer, 8)

      # read compressed message
      @conn.read(@data_buffer, compressed_sz - 8)

      # inflate the results
      # TODO: reuse the current buffer
      uncompressed = Zlib.inflate(@data_buffer.buf)

      @data_buffer = Buffer.new(-1, uncompressed)
    rescue => e
      Aerospike.logger.error("parse result error: #{e}")
      raise e
    end
  else
    begin
      bytes_read = @conn.read(@data_buffer, MSG_TOTAL_HEADER_SIZE - 8, 8)
    rescue => e
      Aerospike.logger.error("parse result error: #{e}")
      raise e
    end
  end

  # A number of these are commented out because we just don't care enough to read
  # that section of the header. If we do care, uncomment and check!
  sz = @data_buffer.read_int64(0)
  header_length = @data_buffer.read(8).ord
  result_code = @data_buffer.read(13).ord & 0xFF
  generation = @data_buffer.read_int32(14)
  expiration = @data_buffer.read_int32(18)
  field_count = @data_buffer.read_int16(26) # almost certainly 0
  op_count = @data_buffer.read_int16(28)
  receive_size = (sz & 0xFFFFFFFFFFFF) - header_length

  # Read remaining message bytes.
  if compressed_sz
    @data_buffer.eat!(MSG_TOTAL_HEADER_SIZE)
  elsif receive_size > 0
    size_buffer_sz(receive_size)

    begin
      @conn.read(@data_buffer, receive_size)
    rescue => e
      Aerospike.logger.error("parse result error: #{e}")
      raise e
    end

  end

  if result_code == 0
    if op_count == 0
      @record = Record.new(@node, @key, nil, generation, expiration)
      return
    end

    @record = parse_record(op_count, field_count, generation, expiration)
    return
  end

  return nil if result_code == Aerospike::ResultCode::KEY_NOT_FOUND_ERROR

  if result_code == Aerospike::ResultCode::FILTERED_OUT
    if @policy.fail_on_filtered_out
      raise Aerospike::Exceptions::Aerospike.new(result_code, nil, [@node])
    end
    return
  end

  if result_code == Aerospike::ResultCode::UDF_BAD_RESPONSE
    begin
      @record = parse_record(op_count, field_count, generation, expiration)
      handle_udf_error(result_code)
    rescue => e
      Aerospike.logger.error("UDF execution error: #{e}")
      raise e
    end
  end

  raise Aerospike::Exceptions::Aerospike.new(result_code, nil, [@node])
end