Class: KafkaSyrup::TopicProducer

Inherits:
Object
  • Object
show all
Includes:
Utils
Defined in:
lib/kafka_syrup/topic_producer.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from Utils

#load_args, #log

Constructor Details

#initialize(topic, *args) ⇒ TopicProducer

Returns a new instance of TopicProducer.



7
8
9
10
11
12
13
14
15
16
17
18
19
 
# File 'lib/kafka_syrup/topic_producer.rb', line 7

def initialize(topic, *args)
  self.topic = topic
  self.opts = args.last || {}

  self.brokers = {}
  self.partitions = {}

  refresh_metadata

  self.partitioner ||= ->(msg) do
    msg.hash % partitions.count
  end
end

Instance Attribute Details

#brokers ⇒ Object

Returns the value of attribute brokers.



5
6
7
 
# File 'lib/kafka_syrup/topic_producer.rb', line 5

def brokers
  @brokers
end

#opts ⇒ Object

Returns the value of attribute opts.



5
6
7
 
# File 'lib/kafka_syrup/topic_producer.rb', line 5

def opts
  @opts
end

#partitioner ⇒ Object

Returns the value of attribute partitioner.



5
6
7
 
# File 'lib/kafka_syrup/topic_producer.rb', line 5

def partitioner
  @partitioner
end

#partitions ⇒ Object

Returns the value of attribute partitions.



5
6
7
 
# File 'lib/kafka_syrup/topic_producer.rb', line 5

def partitions
  @partitions
end

#topic ⇒ Object

Returns the value of attribute topic.



5
6
7
 
# File 'lib/kafka_syrup/topic_producer.rb', line 5

def topic
  @topic
end

Instance Method Details

#refresh_metadata ⇒ Object 



21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
 
# File 'lib/kafka_syrup/topic_producer.rb', line 21

def refresh_metadata
  brokers.each{ |id, broker| broker && broker.socket && broker.socket.close }

  self.brokers.clear
  self.partitions.clear

  meta = KafkaSyrup.get_metadata(topic)

  meta.brokers.each do |broker|
    broker.extend KafkaSyrup::Broker::Communications
    brokers.store broker.node, broker
  end

  meta.topics.first.partitions.each do |partition|
    partitions.store partition.id, partition
  end
end

#send_message(msg) ⇒ Object 



39
40
41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/kafka_syrup/topic_producer.rb', line 39

def send_message(msg)
  id = partitioner.call(msg)

  request = KafkaSyrup::Protocol::ProduceRequest.new
  request.add_topic(topic).add_partition(id).add_message(msg)

  broker_for_id(id).send_request(request)

rescue KafkaResponseErrors::NotLeaderForPartition, SocketReadError
  refresh_metadata

  retry
end