Class: Sbmt::KafkaConsumer::InboxConsumer

Inherits:
BaseConsumer
  • Object
show all
Defined in:
lib/sbmt/kafka_consumer/inbox_consumer.rb

Constant Summary collapse

IDEMPOTENCY_HEADER_NAME = 
"Idempotency-Key"
DEFAULT_SOURCE = 
"KAFKA"

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from BaseConsumer

#consume, #process_batch?

Constructor Details

#initialize ⇒ InboxConsumer

Returns a new instance of InboxConsumer.

Raises:



20
21
22
23
 
# File 'lib/sbmt/kafka_consumer/inbox_consumer.rb', line 20

def initialize
  raise Sbmt::KafkaConsumer::Error, "inbox_item param is not set" if inbox_item_class.blank?
  super
end

Class Method Details

.consumer_klass(inbox_item:, event_name: nil, skip_on_error: nil, name: nil, middlewares: nil, batch_middlewares: nil) ⇒ Object 



12
13
14
15
16
17
18
 
# File 'lib/sbmt/kafka_consumer/inbox_consumer.rb', line 12

def self.consumer_klass(inbox_item:, event_name: nil, skip_on_error: nil, name: nil, middlewares: nil, batch_middlewares: nil)
  # defaults are set in class_attribute definition
  klass = super(skip_on_error: skip_on_error, middlewares: middlewares, batch_middlewares: batch_middlewares)
  klass.inbox_item_class = inbox_item.constantize
  klass.event_name = event_name if event_name
  klass
end

Instance Method Details

#extra_message_attrs(_message) ⇒ Object 



25
26
27
 
# File 'lib/sbmt/kafka_consumer/inbox_consumer.rb', line 25

def extra_message_attrs(_message)
  {}
end