Custom parser class about karafka HOT 8 CLOSED

class MyController < Karafka::BaseController
     self.parser = lambda do |raw_message|
       dynamic_parser.parse(raw_message)
     end
end

class ParserProxy
  class ParserError < Karafka::Errors::ParserError; end

  attr_accessor :parser

  def self.proxy(proto_klass)
    new(proto_klass)
  end

  def initialize(klass)
    @klass = klass
  end

  def parse(msg)
    @klass.decode(msg)
  rescue => e
    raise ParserError
  end
end

# using parser proxy above
class LandingInvitationController < Karafka::BaseController
  self.topic = 'mail.landing_invitation'
  self.parser = ParserProxy.new(Yam::LandingInvitation)

  def perform
    "foo"
  end
end

@klass.decode(Base64.strict_decode64(msg))

class LandingInvitationController < Karafka::BaseController
  self.topic = 'mail.landing_invitation'
  self.worker = LandingInvitationJob

  def perform
  end
end

class LandingInvitationJob < BaseJob
  sidekiq_options :queue => :landing_invitation
  serializer Yam::LandingInvitation

  Contract Yam::LandingInvitation => Any
  def process(payload)
    Karafka.logger.info(payload)
  end

  def on_failure
   #do_something
  end
end

# and the custom worker

class BaseJob < ::SidekiqGlass::Worker
  include Sidekiq::Worker
  include Contracts

  attr_accessor :params

  def self.serializer(klass)
    define_method :parse do |message|
      klass.decode(Base64.strict_decode64(message))
    end
  end

  # @param args [Array] controller params and controller topic
  # @note Arguments are provided in Karafka::BaseController enqueue
  def execute(*args)
    self.params = args.first
    Karafka.logger.info("#{self.class}#execute for #{params}")

    if self.respond_to?(:parse)
      process(parse(params["message"]))
    else
      process(params["message"])
    end
  end

  # What action should be taken when execute method fails
  # With after_failure we can provide reentrancy to this worker
  # @param args [Array] controller params and controller topic
  def after_failure(*args)
    self.params = args.first

    unless self.respond_to?(:on_failure)
      Karafka.logger.warn("#{self.class}#on_failure missing for #{params}")
      return
    end

    Karafka.logger.warn("#{self.class}#on_failure for #{args}")
    self.on_failure
  end
end

class TestController < Karafka::BaseController
  self.interchanger = Base64Interchanger
end

class Base64Interchanger
  class << self
    def load(params)
      Base64.encode64(Marshal.dump(params))
    end

    def parse(params)
      Marshal.load(Base64.decode64(params))
    end
  end
end