Class: WaterDrop::SyncProducer

Inherits:

BaseProducer

• Object
• BaseProducer
• WaterDrop::SyncProducer

Defined in:

lib/water_drop/sync_producer.rb

Overview

Sync producer for messages

Class Method Summary

• .call(message, options) ⇒ Object

  Performs message delivery using deliver_async method.

Class Method Details

.call(message, options) ⇒ Object

Performs message delivery using deliver_async method

Parameters:

• message (String) —

  message that we want to send to Kafka

• options (Hash) —

  options (including topic) for producer

Raises:

• (WaterDrop::Errors::InvalidMessageOptions) —

  raised when message options are somehow invalid and we cannot perform delivery because of that

# File 'lib/water_drop/sync_producer.rb', line 12

def self.call(message, options)
  attempts_count ||= 0
  attempts_count += 1

  validate!(options)
  return unless WaterDrop.config.deliver

  DeliveryBoy.deliver(message, options)
rescue Kafka::Error => error
  graceful_attempt?(attempts_count, message, options, error) ? retry : raise(error)
end