Sha256: 9e97d57d359c2affbb3b646887831f652e13399f519085692dc168934a724044
Contents?: true
Size: 1.7 KB
Versions: 19
Compression:
Stored size: 1.7 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,
offset: kafka_message.offset,
deserializers: topic.deserializers,
partition: kafka_message.partition,
topic: topic.name,
received_at: received_at,
raw_headers: kafka_message.headers,
raw_key: kafka_message.key
)
# Get the raw payload
payload = kafka_message.payload
# And nullify it in the kafka message. This can save a lot of memory when used with
# the Pro Cleaner API
kafka_message.instance_variable_set('@payload', nil)
# Karafka messages cannot be frozen because of the lazy deserialization feature
message = Karafka::Messages::Message.new(payload, metadata)
# Assign message to metadata so we can reverse its relationship if needed
metadata[:message] = message
message
end
end
end
end
end
end
Version data entries
19 entries across 19 versions & 1 rubygems