Class: PlainBufferCodedInputStream

Inherits:
Object
  • Object
show all
Defined in:
lib/tablestore/plain_buffer_coded_input_stream.rb

Instance Method Summary collapse

Constructor Details

#initialize(input_stream) ⇒ PlainBufferCodedInputStream

Returns a new instance of PlainBufferCodedInputStream.



6
7
8
 
# File 'lib/tablestore/plain_buffer_coded_input_stream.rb', line 6

def initialize(input_stream)
  @input_stream = input_stream
end

Instance Method Details

#check_last_tag_was(tag) ⇒ Object 



14
15
16
 
# File 'lib/tablestore/plain_buffer_coded_input_stream.rb', line 14

def check_last_tag_was(tag)
  @input_stream.check_last_tag_was(tag)
end

#get_last_tag ⇒ Object 



18
19
20
 
# File 'lib/tablestore/plain_buffer_coded_input_stream.rb', line 18

def get_last_tag
  @input_stream.get_last_tag
end

#read_column(row_check_sum) ⇒ Object

Raises:



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
159
160
161
162
 
# File 'lib/tablestore/plain_buffer_coded_input_stream.rb', line 128

def read_column(row_check_sum)
  raise TableStoreClientError.new("Expect TAG_CELL but it was " + get_last_tag.to_s) unless check_last_tag_was(TAG_CELL)
  read_tag
  raise TableStoreClientError.new("Expect TAG_CELL_NAME but it was " + get_last_tag.to_s) unless check_last_tag_was(TAG_CELL_NAME)

  cell_check_sum = 0
  column_value = nil
  timestamp = nil
  name_size = @input_stream.read_raw_little_endian32
  column_name = @input_stream.read_utf_string(name_size)
  cell_check_sum = PlainBufferCrc8.crc_string(cell_check_sum, column_name)
  read_tag
  if get_last_tag == TAG_CELL_VALUE
    column_value, cell_check_sum = read_column_value(cell_check_sum)
  end
  # skip CELL_TYPE
  if get_last_tag == TAG_CELL_TYPE
    cell_check_sum = PlainBufferCrc8.crc_int8(cell_check_sum, cell_type)
    read_tag
  end
  if get_last_tag == TAG_CELL_TIMESTAMP
    timestamp = @input_stream.read_int64
    cell_check_sum = PlainBufferCrc8.crc_int64(cell_check_sum, timestamp)
    read_tag
  end
  if get_last_tag == TAG_CELL_CHECKSUM
    check_sum = @input_stream.read_raw_byte.ord
    raise TableStoreClientError.new("Checksum mismatch. expected:" + check_sum.to_s + ",actual:" + cell_check_sum.to_s) if check_sum != cell_check_sum
    read_tag
  else
    raise TableStoreClientError.new("Expect TAG_CELL_CHECKSUM but it was " + get_last_tag.to_s)
  end
  row_check_sum = PlainBufferCrc8.crc_int8(row_check_sum, cell_check_sum)
  return column_name, column_value, timestamp, row_check_sum
end

#read_column_value(cell_check_sum) ⇒ Object

Raises:



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
 
# File 'lib/tablestore/plain_buffer_coded_input_stream.rb', line 57

def read_column_value(cell_check_sum)
  raise TableStoreClientError.new("Expect TAG_CELL_VALUE but it was " + get_last_tag.to_s) unless check_last_tag_was(TAG_CELL_VALUE)
  @input_stream.read_raw_little_endian32
  column_type = @input_stream.read_raw_byte.ord
  if column_type == VT_INTEGER
    int64_value = @input_stream.read_int64
    cell_check_sum = PlainBufferCrc8.crc_int8(cell_check_sum, VT_INTEGER)
    cell_check_sum = PlainBufferCrc8.crc_int64(cell_check_sum, int64_value)
    read_tag
    return int64_value, cell_check_sum
  elsif column_type == VT_STRING
    value_size = @input_stream.read_int32
    string_value = @input_stream.read_utf_string(value_size)
    cell_check_sum = PlainBufferCrc8.crc_int8(cell_check_sum, VT_STRING)
    cell_check_sum = PlainBufferCrc8.crc_int32(cell_check_sum, value_size)
    cell_check_sum =PlainBufferCrc8.crc_string(cell_check_sum, string_value)
    read_tag
    return string_value, cell_check_sum
  elsif column_type == VT_BLOB
    value_size = @input_stream.read_int32
    binary_value = @input_stream.read_bytes(value_size)
    cell_check_sum = PlainBufferCrc8.crc_int8(cell_check_sum, VT_BLOB)
    cell_check_sum = PlainBufferCrc8.crc_int32(cell_check_sum, value_size)
    cell_check_sum = PlainBufferCrc8.crc_string(cell_check_sum, binary_value)
    read_tag
    return binary_value, cell_check_sum
  elsif column_type == VT_BOOLEAN
    bool_value = @input_stream.read_boolean
    cell_check_sum = PlainBufferCrc8.crc_int8(cell_check_sum, VT_BOOLEAN)
    cell_check_sum = PlainBufferCrc8.crc_int8(cell_check_sum, bool_value ? 1 : 0)
    read_tag
    return bool_value, cell_check_sum
  elsif column_type == VT_DOUBLE
    double_int = @input_stream.read_int64
    cell_check_sum = PlainBufferCrc8.crc_int8(cell_check_sum, VT_DOUBLE)
    cell_check_sum = PlainBufferCrc8.crc_int64(cell_check_sum, double_int)
    read_tag

    if SYS_BITS == 64
        double_value, = [double_int].pack('q').unpack('d')
    else
        double_value, = [double_int].pack('l').unpack('d')
    end
    return double_value, cell_check_sum
  else
    raise TableStoreClientError.new("Unsupported column type: " + column_type.str)
  end
end

#read_header ⇒ Object 



22
23
24
 
# File 'lib/tablestore/plain_buffer_coded_input_stream.rb', line 22

def read_header
  @input_stream.read_int32
end

#read_primary_key_column(row_check_sum) ⇒ Object

Raises:



106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
 
# File 'lib/tablestore/plain_buffer_coded_input_stream.rb', line 106

def read_primary_key_column(row_check_sum)
  raise TableStoreClientError.new("Expect TAG_CELL but it was " + get_last_tag.to_s) unless check_last_tag_was(TAG_CELL)
  read_tag
  raise TableStoreClientError.new("Expect TAG_CELL_NAME but it was " + get_last_tag.to_s) unless check_last_tag_was(TAG_CELL_NAME)
  cell_check_sum = 0
  name_size = @input_stream.read_raw_little_endian32
  column_name = @input_stream.read_utf_string(name_size)
  cell_check_sum = PlainBufferCrc8.crc_string(cell_check_sum, column_name)
  read_tag
  raise TableStoreClientError.new("Expect TAG_CELL_VALUE but it was " + get_last_tag.to_s) unless check_last_tag_was(TAG_CELL_VALUE)
  primary_key_value, cell_check_sum = read_primary_key_value(cell_check_sum)
  if @input_stream.get_last_tag == TAG_CELL_CHECKSUM
    check_sum = @input_stream.read_raw_byte.ord
    raise TableStoreClientError.new("Checksum mismatch. expected:" + check_sum.to_s + ",actual:" + cell_check_sum.to_s) if check_sum != cell_check_sum
    read_tag
  else
    raise TableStoreClientError.new("Expect TAG_CELL_CHECKSUM but it was " + get_last_tag.to_s)
  end
  row_check_sum = PlainBufferCrc8.crc_int8(row_check_sum, cell_check_sum)
  return column_name, primary_key_value, row_check_sum
end

#read_primary_key_value(cell_check_sum) ⇒ Object

Raises:



26
27
28
29
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
 
# File 'lib/tablestore/plain_buffer_coded_input_stream.rb', line 26

def read_primary_key_value(cell_check_sum)
  raise TableStoreClientError.new("Expect TAG_CELL_VALUE but it was " + get_last_tag.to_s) unless check_last_tag_was(TAG_CELL_VALUE)
  @input_stream.read_raw_little_endian32
  column_type = @input_stream.read_raw_byte.ord
  if column_type == VT_INTEGER
    int64_value = @input_stream.read_int64
    cell_check_sum = PlainBufferCrc8.crc_int8(cell_check_sum, VT_INTEGER)
    cell_check_sum = PlainBufferCrc8.crc_int64(cell_check_sum, int64_value)
    read_tag
    return int64_value, cell_check_sum
  elsif column_type == VT_STRING
    value_size = @input_stream.read_int32
    string_value = @input_stream.read_utf_string(value_size)
    cell_check_sum = PlainBufferCrc8.crc_int8(cell_check_sum, VT_STRING)
    cell_check_sum = PlainBufferCrc8.crc_int32(cell_check_sum, value_size)
    cell_check_sum = PlainBufferCrc8.crc_string(cell_check_sum, string_value)
    read_tag
    return string_value, cell_check_sum
  elsif column_type == VT_BLOB
    value_size = @input_stream.read_int32
    binary_value = @input_stream.read_bytes(value_size)
    cell_check_sum = PlainBufferCrc8.crc_int8(cell_check_sum, VT_BLOB)
    cell_check_sum = PlainBufferCrc8.crc_int32(cell_check_sum, value_size)
    cell_check_sum = PlainBufferCrc8.crc_string(cell_check_sum, binary_value)
    read_tag
    return [binary_value], cell_check_sum
  else
    raise TableStoreClientError.new("Unsupported primary key type:" + column_type.to_s)
  end
end

#read_row ⇒ Object

Raises:



200
201
202
203
204
 
# File 'lib/tablestore/plain_buffer_coded_input_stream.rb', line 200

def read_row
  raise TableStoreClientError.new("Invalid header from plain buffer.") if read_header != HEADER
  read_tag
  read_row_without_header
end

#read_row_without_header ⇒ Object

Raises:



164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
 
# File 'lib/tablestore/plain_buffer_coded_input_stream.rb', line 164

def read_row_without_header
  row_check_sum = 0
  primary_key = {}
  attributes = {}

  raise TableStoreClientError.new("Expect TAG_ROW_PK but it was " + get_last_tag.to_s) unless check_last_tag_was(TAG_ROW_PK)

  read_tag
  while check_last_tag_was(TAG_CELL)
    name, value, row_check_sum = read_primary_key_column(row_check_sum)
    primary_key[name] = value
  end
  if check_last_tag_was(TAG_ROW_DATA)
    read_tag
    while check_last_tag_was(TAG_CELL)
      column_name, column_value, timestamp, row_check_sum = read_column(row_check_sum)
      attributes[column_name] = column_value
    end
  end
  if check_last_tag_was(TAG_DELETE_ROW_MARKER)
    read_tag
    row_check_sum = PlainBufferCrc8.crc_int8(row_check_sum, 1)
  else
    row_check_sum = PlainBufferCrc8.crc_int8(row_check_sum, 0)
  end

  if check_last_tag_was(TAG_ROW_CHECKSUM)
    check_sum = @input_stream.read_raw_byte.ord
    raise TableStoreClientError.new("Checksum is mismatch.") if check_sum != row_check_sum
    read_tag
  else
    raise TableStoreClientError.new("Expect TAG_ROW_CHECKSUM but it was " + get_last_tag.to_s)
  end
  return primary_key, attributes
end

#read_rows ⇒ Object 



206
207
208
209
210
211
212
213
214
215
 
# File 'lib/tablestore/plain_buffer_coded_input_stream.rb', line 206

def read_rows
  raise TableStoreClientError("Invalid header from plain buffer.") if read_header != HEADER
  read_tag
  row_list = []
  while !@input_stream.is_at_end?
    pk, attr = read_row_without_header
    row_list << {"primary_key"=>pk, "attribute_columns"=> attr}
  end
  row_list
end

#read_tag ⇒ Object 



10
11
12
 
# File 'lib/tablestore/plain_buffer_coded_input_stream.rb', line 10

def read_tag
  @input_stream.read_tag
end