Sha256: 648be4c6ed5e745461edcd8c2477854706ad33fbfa04e5f37bd23707df40d411
Contents?: true
Size: 1.33 KB
Versions: 9
Compression:
Stored size: 1.33 KB
Contents
# frozen_string_literal: true
module Karafka
module Messages
# Builders encapsulate logic related to creating messages related objects.
module Builders
# Builder of a single message based on raw rdkafka message.
module Message
class << self
# @param kafka_message [Rdkafka::Consumer::Message] raw fetched message
# @param topic [Karafka::Routing::Topic] topic for which this message was fetched
# @param received_at [Time] moment when we've received the message
# @return [Karafka::Messages::Message] message object with payload and metadata
def call(kafka_message, topic, received_at)
metadata = Karafka::Messages::Metadata.new(
timestamp: kafka_message.timestamp,
headers: kafka_message.headers,
key: kafka_message.key,
offset: kafka_message.offset,
deserializer: topic.deserializer,
partition: kafka_message.partition,
topic: topic.name,
received_at: received_at
).freeze
# Karafka messages cannot be frozen because of the lazy deserialization feature
Karafka::Messages::Message.new(
kafka_message.payload,
metadata
)
end
end
end
end
end
end
Version data entries
9 entries across 9 versions & 1 rubygems