Class: Aerospike::StreamCommand

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

Overview

:nodoc:

Direct Known Subclasses

QueryCommand, ScanCommand, ScanPartitionCommand

Instance Method Summary collapse

Methods inherited from MultiCommand

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

Methods inherited from Command

#execute, #initialize, #set_delete, #set_exists, #set_operate, #set_query, #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
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
 
# 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::PARTITION_UNAVAILABLE
      # 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, nil, [@node])
    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 cmd is the end marker of the response, do not proceed further
    if (info3 & INFO3_PARTITION_DONE) != 0
      # When an error code is received, mark partition as unavailable
      # for the current round. Unavailable partitions will be retried
      # in the next round. Generation is overloaded as partitionId.
      if result_code != 0
        @tracker&.partition_unavailable(@node_partitions, generation)
      end

      next
    end

    next unless 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

    # UDF results do not return a key
    @tracker&.set_last(@node_partitions, key, key.bval) if key
  end # while

  true
end