Sha256: 87ed352e754332421ad1ab50ae1ed9fd0575735422b68aea98b4b4e0548bb495
Contents?: true
Size: 734 Bytes
Versions: 11
Compression:
Stored size: 734 Bytes
Contents
# frozen_string_literal: true
module Deimos
module Backends
# Backend which produces to Kafka via an async producer.
class KafkaAsync < Deimos::PublishBackend
include Phobos::Producer
# :nodoc:
def self.execute(producer_class:, messages:)
Deimos.instrument(
'produce',
producer: producer_class,
topic: producer_class.topic,
payloads: messages.map(&:payload)
) do
producer.async_publish_list(messages.map(&:encoded_hash))
Deimos.config.metrics&.increment(
'publish',
tags: %W(status:success topic:#{producer_class.topic}),
by: messages.size
)
end
end
end
end
end
Version data entries
11 entries across 11 versions & 2 rubygems