Class: Sbmt::KafkaConsumer::Serialization::ProtobufDeserializer

Inherits:
BaseDeserializer
  • Object
show all
Defined in:
lib/sbmt/kafka_consumer/serialization/protobuf_deserializer.rb

Instance Attribute Summary collapse

Attributes inherited from BaseDeserializer

#skip_decoding_error

Instance Method Summary collapse

Constructor Details

#initialize(message_decoder_klass:, skip_decoding_error: false) ⇒ ProtobufDeserializer

Returns a new instance of ProtobufDeserializer.



11
12
13
14
15
 
# File 'lib/sbmt/kafka_consumer/serialization/protobuf_deserializer.rb', line 11

def initialize(message_decoder_klass:, skip_decoding_error: false)
  super(skip_decoding_error: skip_decoding_error)

  @message_decoder = message_decoder_klass.constantize
end

Instance Attribute Details

#message_decoder ⇒ Object (readonly)

Returns the value of attribute message_decoder.



9
10
11
 
# File 'lib/sbmt/kafka_consumer/serialization/protobuf_deserializer.rb', line 9

def message_decoder
  @message_decoder
end

Instance Method Details

#call(message) ⇒ Object 



17
18
19
20
21
22
23
 
# File 'lib/sbmt/kafka_consumer/serialization/protobuf_deserializer.rb', line 17

def call(message)
  message_decoder.decode(message.raw_payload)
rescue Google::Protobuf::ParseError, ArgumentError => e
  raise Sbmt::KafkaConsumer::SkipUndeserializableMessage, "cannot decode message: #{e.message}, payload: #{message.raw_payload}" if skip_decoding_error

  raise
end