Class: Kafka::Protocol::Message

26
27
28
29
30
31
32
33
34

# File 'lib/kafka/protocol/message.rb', line 26
def initialize(value:, key: nil, create_time: Time.now, codec_id: 0, offset: -1)
 @key = key
 @value = value
 @codec_id = codec_id
 @offset = offset
 @create_time = create_time
 @bytesize = @key.to_s.bytesize + @value.to_s.bytesize
end

24
25
26

# File 'lib/kafka/protocol/message.rb', line 24
def bytesize
 @bytesize
end

22
23
24

# File 'lib/kafka/protocol/message.rb', line 22
def codec_id
 @codec_id
end

24
25
26

# File 'lib/kafka/protocol/message.rb', line 24
def create_time
 @create_time
end

22
23
24

# File 'lib/kafka/protocol/message.rb', line 22
def key
 @key
end

22
23
24

# File 'lib/kafka/protocol/message.rb', line 22
def offset
 @offset
end

22
23
24

# File 'lib/kafka/protocol/message.rb', line 22
def value
 @value
end

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

# File 'lib/kafka/protocol/message.rb', line 66
def self.decode(decoder)
 offset = decoder.int64
 message_decoder = Decoder .from_string (decoder.bytes)
 _crc = message_decoder.int32
 magic_byte = message_decoder.int8
 attributes = message_decoder.int8
 # The magic byte indicates the message format version. There are situations
 # where an old message format can be returned from a newer version of Kafka,
 # because old messages are not necessarily rewritten on upgrades.
 case magic_byte
 when 0
 # No timestamp in the pre-0.10 message format.
 timestamp = nil
 when 1
 timestamp = message_decoder.int64
 # If the timestamp is set to zero, it's because the message has been upgraded
 # from the Kafka 0.9 disk format to the Kafka 0.10 format. The former didn't
 # have a timestamp attribute, so we'll just set the timestamp to nil.
 timestamp = nil if timestamp.zero?
 else
 raise Kafka ::Error , "Invalid magic byte: #{magic_byte}"
 end
 key = message_decoder.bytes
 value = message_decoder.bytes
 # The codec id is encoded in the three least significant bits of the
 # attributes.
 codec_id = attributes & 0b111
 # The timestamp will be nil if the message was written in the Kafka 0.9 log format.
 create_time = timestamp && Time.at(timestamp / 1000.0)
 new(key: key, value: value, codec_id: codec_id, offset: offset, create_time: create_time)
end

43
44
45
46
47
48

# File 'lib/kafka/protocol/message.rb', line 43
def ==(other)
 @key == other.key &&
 @value == other.value &&
 @codec_id == other.codec_id &&
 @offset == other.offset
end

50
51
52

# File 'lib/kafka/protocol/message.rb', line 50
def compressed?
 @codec_id != 0
end

55
56
57
58
59
60
61
62
63
64

# File 'lib/kafka/protocol/message.rb', line 55
def decompress
 codec = Compression .find_codec_by_id (@codec_id)
 # For some weird reason we need to cut out the first 20 bytes.
 data = codec.decompress(value)
 message_set_decoder = Decoder .from_string (data)
 message_set = MessageSet .decode (message_set_decoder)
 correct_offsets(message_set.messages)
end

36
37
38
39
40
41

# File 'lib/kafka/protocol/message.rb', line 36
def encode(encoder)
 data = encode_with_crc
 encoder.write_int64(offset)
 encoder.write_bytes(data)
end

110
111
112

# File 'lib/kafka/protocol/message.rb', line 110
def headers
 {}
end

106
107
108

# File 'lib/kafka/protocol/message.rb', line 106
def is_control_record
 false
end