Class: Aerospike::ReadCommand

Inherits:
SingleCommand show all
Defined in:
lib/aerospike/command/read_command.rb

Overview

:nodoc:

Direct Known Subclasses

ExecuteCommand, OperateCommand

Constant Summary collapse

BIN_NAME_ENCODING = 
'utf-8'

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods inherited from Command

#execute, #set_delete, #set_exists, #set_operate, #set_read, #set_read_for_key_only, #set_read_header, #set_scan, #set_touch, #set_udf, #set_write, #write_bins

Constructor Details

#initialize(cluster, policy, key, bin_names) ⇒ ReadCommand

Returns a new instance of ReadCommand.



33
34
35
36
37
38
39
40
 
# File 'lib/aerospike/command/read_command.rb', line 33

def initialize(cluster, policy, key, bin_names)
  super(cluster, key)

  @bin_names = bin_names
  @policy = policy

  self
end

Instance Attribute Details

#policy ⇒ Object (readonly)

Returns the value of attribute policy.



31
32
33
 
# File 'lib/aerospike/command/read_command.rb', line 31

def policy
  @policy
end

#record ⇒ Object (readonly)

Returns the value of attribute record.



31
32
33
 
# File 'lib/aerospike/command/read_command.rb', line 31

def record
  @record
end

Instance Method Details

#handle_udf_error(result_code) ⇒ Object

Raises:



111
112
113
114
115
 
# File 'lib/aerospike/command/read_command.rb', line 111

def handle_udf_error(result_code)
  ret = @record.bins['FAILURE']
  raise Aerospike::Exceptions::Aerospike.new(result_code, ret) if ret
  raise Aerospike::Exceptions::Aerospike.new(result_code)
end

#parse_record(op_count, field_count, generation, expiration) ⇒ Object 



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
146
147
148
149
150
151
152
153
154
155
156
157
158
 
# File 'lib/aerospike/command/read_command.rb', line 117

def parse_record(op_count, field_count, generation, expiration)
  bins = op_count > 0 ? {} : nil
  receive_offset = 0
  single_bin_value = (!(OperatePolicy === policy) || policy.record_bin_multiplicity == RecordBinMultiplicity::SINGLE)

  # There can be fields in the response (setname etc).
  # But for now, ignore them. Expose them to the API if needed in the future.
  if field_count > 0
    # Just skip over all the fields
    i = 0
    while i < field_count
      field_size = @data_buffer.read_int32(receive_offset)
      receive_offset += (4 + field_size)
      i = i.succ
    end
  end

  i = 0
  while i < op_count
    op_size = @data_buffer.read_int32(receive_offset)
    particle_type = @data_buffer.read(receive_offset+5).ord
    name_size = @data_buffer.read(receive_offset+7).ord
    name = @data_buffer.read(receive_offset+8, name_size).force_encoding(BIN_NAME_ENCODING)
    receive_offset += 4 + 4 + name_size

    particle_bytes_size = op_size - (4 + name_size)
    value = Aerospike.bytes_to_particle(particle_type, @data_buffer, receive_offset, particle_bytes_size)
    receive_offset += particle_bytes_size

    if single_bin_value || !bins.has_key?(name)
      bins[name] = value
    elsif (prev = bins[name]).kind_of?(OpResults)
      prev << value
    else
      bins[name] = OpResults.new << prev << value
    end

    i = i.succ
  end

  Record.new(@node, @key, bins, generation, expiration)
end

#parse_result ⇒ Object

Raises:



46
47
48
49
50
51
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
 
# File 'lib/aerospike/command/read_command.rb', line 46

def parse_result
  # Read header.
  begin
    @conn.read(@data_buffer, MSG_TOTAL_HEADER_SIZE)
  rescue => e
    Aerospike.logger.error("parse result error: #{e}")
    raise e
  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 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)
    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)
end

#write_buffer ⇒ Object 



42
43
44
 
# File 'lib/aerospike/command/read_command.rb', line 42

def write_buffer
  set_read(@policy, @key, @bin_names)
end