Class: Aerospike::StreamCommand

Inherits:
MultiCommand show all
Defined in:
lib/aerospike/query/stream_command.rb

Overview

:nodoc:

Direct Known Subclasses

QueryCommand, ScanCommand

Instance Method Summary collapse

Methods inherited from MultiCommand

#compressed?, #get_node, #initialize, #parse_key, #parse_record, #parse_result, #read_bytes, #stop, #valid?

Methods inherited from Command

#execute, #initialize, #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

This class inherits a constructor from Aerospike::MultiCommand

Instance Method Details

#parse_group(receive_size) ⇒ Object 



30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
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
 
# File 'lib/aerospike/query/stream_command.rb', line 30

def parse_group(receive_size)
  @data_offset = 0

  while @data_offset < receive_size
    read_bytes(MSG_REMAINING_HEADER_SIZE)
    result_code = @data_buffer.read(5).ord & 0xFF

    # The only valid server return codes are "ok" and "not found".
    # If other return codes are received, then abort the batch.
    case result_code
    when Aerospike::ResultCode::OK
      # noop
    when Aerospike::ResultCode::KEY_NOT_FOUND_ERROR
      # consume the rest of the input buffer from the socket
      read_bytes(receive_size - @data_offset) if @data_offset < receive_size
      return nil
    else
      raise Aerospike::Exceptions::Aerospike.new(result_code)
    end

    info3 = @data_buffer.read(3).ord

    # If cmd is the end marker of the response, do not proceed further
    return false if (info3 & INFO3_LAST) == INFO3_LAST

    generation = @data_buffer.read_int32(6)
    expiration = @data_buffer.read_int32(10)
    field_count = @data_buffer.read_int16(18)
    op_count = @data_buffer.read_int16(20)
    key = parse_key(field_count)

    if result_code == 0
      if @recordset.active?
        @recordset.records.enq(parse_record(key, op_count, generation, expiration))
      else
        expn = @recordset.is_scan? ? SCAN_TERMINATED_EXCEPTION : QUERY_TERMINATED_EXCEPTION
        raise expn
      end
    end
  end # while

  true
end