Sha256: d4fc0af2584665d4f67153a789fae62b90229faefca0db7759742501a0ba66b3
Contents?: true
Size: 1.25 KB
Versions: 60
Compression:
Stored size: 1.25 KB
Contents
# frozen_string_literal: true
module Karafka
module Messages
module Builders
# Builder for creating message batch instances.
module Messages
class << self
# Creates messages batch with messages inside based on the incoming messages and the
# topic from which it comes.
#
# @param messages [Array<Karafka::Messages::Message>] karafka messages array
# @param topic [Karafka::Routing::Topic] topic for which we're received messages
# @param partition [Integer] partition of those messages
# @param received_at [Time] moment in time when the messages were received
# @return [Karafka::Messages::Messages] messages batch object
def call(messages, topic, partition, received_at)
# We cannot freeze the batch metadata because it is altered with the processed_at time
# prior to the consumption. It is being frozen there
metadata = BatchMetadata.call(
messages,
topic,
partition,
received_at
)
Karafka::Messages::Messages.new(
messages,
metadata
).freeze
end
end
end
end
end
end
Version data entries
60 entries across 60 versions & 1 rubygems