Sha256: dacef660f8058d097820ee78b23d4620b5e2df112b8c6a79bfdcc3a3847a15b6
Contents?: true
Size: 1.88 KB
Versions: 10
Compression:
Stored size: 1.88 KB
Contents
# frozen_string_literal: true module WaterDrop module Instrumentation module Callbacks # Creates a callable that we want to run upon each message delivery or failure # # @note We don't have to provide client_name here as this callback is per client instance class Delivery # @param producer_id [String] id of the current producer # @param monitor [WaterDrop::Instrumentation::Monitor] monitor we are using def initialize(producer_id, monitor) @producer_id = producer_id @monitor = monitor end # Emits delivery details to the monitor # @param delivery_report [Rdkafka::Producer::DeliveryReport] delivery report def call(delivery_report) if delivery_report.error.to_i.positive? instrument_error(delivery_report) else instrument_acknowledged(delivery_report) end end private # @param delivery_report [Rdkafka::Producer::DeliveryReport] delivery report def instrument_error(delivery_report) @monitor.instrument( 'error.occurred', caller: self, error: ::Rdkafka::RdkafkaError.new(delivery_report.error), producer_id: @producer_id, offset: delivery_report.offset, partition: delivery_report.partition, topic: delivery_report.topic_name, type: 'librdkafka.dispatch_error' ) end # @param delivery_report [Rdkafka::Producer::DeliveryReport] delivery report def instrument_acknowledged(delivery_report) @monitor.instrument( 'message.acknowledged', producer_id: @producer_id, offset: delivery_report.offset, partition: delivery_report.partition, topic: delivery_report.topic_name ) end end end end end
Version data entries
10 entries across 10 versions & 1 rubygems