Sha256: cd3bd036740c6d96eb2c3fb1418b118540b3e9cf44d6171106a34d6bbec72e0f
Contents?: true
Size: 926 Bytes
Versions: 7
Compression:
Stored size: 926 Bytes
Contents
# frozen_string_literal: true
# WaterDrop library
module WaterDrop
# Async producer for messages
class AsyncProducer < BaseProducer
# Performs message delivery using deliver_async method
# @param message [String] message that we want to send to Kafka
# @param options [Hash] options (including topic) for producer
# @raise [WaterDrop::Errors::InvalidMessageOptions] raised when message options are
# somehow invalid and we cannot perform delivery because of that
def self.call(message, options)
attempts_count ||= 0
attempts_count += 1
validate!(options)
return unless WaterDrop.config.deliver
d_method = WaterDrop.config.raise_on_buffer_overflow ? :deliver_async! : :deliver_async
DeliveryBoy.send(d_method, message, **options)
rescue Kafka::Error => e
graceful_attempt?(attempts_count, message, options, e) ? retry : raise(e)
end
end
end
Version data entries
7 entries across 7 versions & 1 rubygems