Sha256: 8975edfef9d5b5b44a5b8baea673855ddb414db9b2734c7efc69fce32303c2d1
Contents?: true
Size: 925 Bytes
Versions: 38
Compression:
Stored size: 925 Bytes
Contents
# frozen_string_literal: true
module Deimos
module Backends
# Default backend to produce to Kafka.
class Kafka < Base
include Phobos::Producer
# Shut down the producer if necessary.
# @return [void]
def self.shutdown_producer
producer.sync_producer_shutdown if producer.respond_to?(:sync_producer_shutdown)
producer.kafka_client&.close
end
# :nodoc:
def self.execute(producer_class:, messages:)
Deimos.instrument(
'produce',
producer: producer_class,
topic: producer_class.topic,
payloads: messages.map(&:payload)
) do
producer.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
38 entries across 38 versions & 1 rubygems