Class: KafkaSyrup::Protocol::Message

Inherits:
Object
  • Object
show all
Includes:
Utils
Defined in:
lib/kafka_syrup/protocol/message.rb

Constant Summary collapse

MAGIC_BYTE = 
0

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from Utils

#load_args, #log

Constructor Details

#initialize(*args) ⇒ Message

Returns a new instance of Message.



12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
 
# File 'lib/kafka_syrup/protocol/message.rb', line 12

def initialize(*args)
  if (io = args.first).respond_to?(:read)
    opts = { check_crc: false }
    opts.merge(args.last) if args.last.is_a?(Hash)
    load_args(opts)

    crc = E.read_int32(io)
    E.read_int8(io)     # Magic Byte (ignored)
    E.read_int8(io)     # Compression (ignored)
    self.key = E.read_bytes(io)
    self.value = E.read_bytes(io)
    # TODO Verify CRC
  end

  load_args(*args)
end

Instance Attribute Details

#key ⇒ Object

Returns the value of attribute key.



10
11
12
 
# File 'lib/kafka_syrup/protocol/message.rb', line 10

def key
  @key
end

#offset ⇒ Object

Returns the value of attribute offset.



10
11
12
 
# File 'lib/kafka_syrup/protocol/message.rb', line 10

def offset
  @offset
end

#value ⇒ Object

Returns the value of attribute value.



10
11
12
 
# File 'lib/kafka_syrup/protocol/message.rb', line 10

def value
  @value
end

Instance Method Details

#==(obj) ⇒ Object 



53
54
55
 
# File 'lib/kafka_syrup/protocol/message.rb', line 53

def ==(obj)
  obj.encode == encode
end

#crc ⇒ Object 



45
46
47
 
# File 'lib/kafka_syrup/protocol/message.rb', line 45

def crc
  @crc ||= Zlib.crc32(encoded)
end

#encode ⇒ Object 



29
30
31
32
33
34
 
# File 'lib/kafka_syrup/protocol/message.rb', line 29

def encode
  [
    E.write_int32(crc),
    encoded
  ].join
end

#encoded ⇒ Object 



36
37
38
39
40
41
42
43
 
# File 'lib/kafka_syrup/protocol/message.rb', line 36

def encoded
  @encoded ||= [
    E.write_int8(MAGIC_BYTE),
    E.write_int8(0), # Currently don't support compression
    E.write_bytes(key),
    E.write_bytes(value)
  ].join
end