Sha256: 58ad664f568031b2543143ad2066c4e789893da5ee0598c6bf912c29b1b05fad
Contents?: true
Size: 779 Bytes
Versions: 19
Compression:
Stored size: 779 Bytes
Contents
# frozen_string_literal: true
require "google/protobuf"
module Sbmt
module KafkaConsumer
module Serialization
class ProtobufDeserializer < BaseDeserializer
attr_reader :message_decoder
def initialize(message_decoder_klass:, skip_decoding_error: false)
super(skip_decoding_error: skip_decoding_error)
@message_decoder = message_decoder_klass.constantize
end
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
end
end
end
end
Version data entries
19 entries across 19 versions & 1 rubygems