Sha256: 887b14157208ef535a6bd5793677e02214cd7b0890d6b0cfb7786b7ce8e9aec9
Contents?: true
Size: 1.57 KB
Versions: 6
Compression:
Stored size: 1.57 KB
Contents
# frozen_string_literal: true
module Sbmt
module KafkaProducer
module Config
class Producer < Anyway::Config
class << self
# Make it possible to access a singleton config instance
# via class methods (i.e., without explicitly calling `instance`)
delegate_missing_to :instance
def coerce_to(struct)
lambda do |raw_attrs|
struct.new(**raw_attrs)
rescue Dry::Types::SchemaError => e
raise_validation_error "cannot parse #{struct}: #{e.message}"
end
end
private
# Returns a singleton config instance
def instance
@instance ||= new
end
end
config_name :kafka_producer
attr_config :ignore_kafka_error, :deliver, :wait_on_queue_full,
:max_payload_size, :max_wait_timeout, :wait_timeout,
:wait_on_queue_full_timeout,
auth: {}, kafka: {},
logger_class: "::Sbmt::KafkaProducer::Logger",
metrics_listener_class: "::Sbmt::KafkaProducer::Instrumentation::YabedaMetricsListener"
coerce_types ignore_kafka_error: :boolean,
deliver: :boolean, wait_on_queue_full: :boolean,
max_payload_size: :integer, max_wait_timeout: :integer,
wait_timeout: :float, wait_on_queue_full_timeout: :float
coerce_types kafka: coerce_to(Kafka)
coerce_types auth: coerce_to(Auth)
def to_kafka_options
kafka.to_kafka_options
.merge(auth.to_kafka_options)
end
end
end
end
end
Version data entries
6 entries across 6 versions & 1 rubygems