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 with payload and message metadata.

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
32
33
 
# File 'lib/karafka/params/builders/params.rb', line 16

def from_kafka_message(kafka_message, topic)
  metadata = Karafka::Params::Metadata.new(
    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: topic.name
  ).freeze

  Karafka::Params::Params.new(
    kafka_message.value,
    metadata
  )
end