Module: Karafka::Params::Builders::Params

Defined in:
lib/karafka/params/builders/params.rb

Overview

Builder for params

Class Method Summary collapse

Class Method Details

.from_kafka_message(kafka_message, topic) ⇒ Karafka::Params::Params

Returns params object.

Parameters:

  • kafka_message (Kafka::FetchedMessage) —

    message fetched from Kafka

  • topic (Karafka::Routing::Topic) —

    topic for which this message was fetched

Returns:



16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
 
# File 'lib/karafka/params/builders/params.rb', line 16

def from_kafka_message(kafka_message, topic)
  Karafka::Params::Params
    .new
    .merge!(
      'create_time' => kafka_message.create_time,
      'headers' => kafka_message.headers || {},
      'is_control_record' => kafka_message.is_control_record,
      'key' => kafka_message.key,
      'offset' => kafka_message.offset,
      'deserializer' => topic.deserializer,
      'partition' => kafka_message.partition,
      'receive_time' => Time.now,
      'topic' => kafka_message.topic,
      'payload' => kafka_message.value
    )
end