Class: Kafka::Protocol::Message

Inherits:
Object
  • Object
show all
Defined in:
lib/kafka/protocol/message.rb

Overview

## API Specification

Message => Crc MagicByte Attributes Timestamp Key Value
    Crc => int32
    MagicByte => int8
    Attributes => int8
    Timestamp => int64, in ms
    Key => bytes
    Value => bytes

Constant Summary collapse

MAGIC_BYTE = 
1

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(value:, key: nil, create_time: Time.now, codec_id: 0, offset: -1)) ⇒ Message

Returns a new instance of 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

Instance Attribute Details

#bytesize ⇒ Object (readonly)

Returns the value of attribute bytesize.



24
25
26
 
# File 'lib/kafka/protocol/message.rb', line 24

def bytesize
  @bytesize
end

#codec_id ⇒ Object (readonly)

Returns the value of attribute codec_id.



22
23
24
 
# File 'lib/kafka/protocol/message.rb', line 22

def codec_id
  @codec_id
end

#create_time ⇒ Object (readonly)

Returns the value of attribute create_time.



24
25
26
 
# File 'lib/kafka/protocol/message.rb', line 24

def create_time
  @create_time
end

#key ⇒ Object (readonly)

Returns the value of attribute key.



22
23
24
 
# File 'lib/kafka/protocol/message.rb', line 22

def key
  @key
end

#offset ⇒ Object (readonly)

Returns the value of attribute offset.



22
23
24
 
# File 'lib/kafka/protocol/message.rb', line 22

def offset
  @offset
end

#value ⇒ Object (readonly)

Returns the value of attribute value.



22
23
24
 
# File 'lib/kafka/protocol/message.rb', line 22

def value
  @value
end

Class Method Details

.decode(decoder) ⇒ Object 



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

Instance Method Details

#==(other) ⇒ Object 



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

#compressed? ⇒ Boolean

Returns:

  • (Boolean) 


50
51
52
 
# File 'lib/kafka/protocol/message.rb', line 50

def compressed?
  @codec_id != 0
end

#decompress ⇒ Array<Kafka::Protocol::Message>

Returns:



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

#encode(encoder) ⇒ Object 



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

#headers ⇒ Object 



110
111
112
 
# File 'lib/kafka/protocol/message.rb', line 110

def headers
  {}
end

#is_control_record ⇒ Object

Ensure the backward compatibility of Message format from Kafka 0.11.x



106
107
108
 
# File 'lib/kafka/protocol/message.rb', line 106

def is_control_record
  false
end