Class: KafkaSession
- Inherits:
-
Object
show all
- Defined in:
- lib/kafka_session.rb,
lib/kafka_session/message.rb,
lib/kafka_session/version.rb,
lib/kafka_session/producer.rb,
lib/kafka_session/mock_producer.rb,
lib/kafka_session/delivered_message.rb
Defined Under Namespace
Classes: DeliveredMessage, Message, MockProducer, Producer
Constant Summary
collapse
- VERSION =
"0.5.0"
Class Attribute Summary collapse
Instance Attribute Summary collapse
Class Method Summary
collapse
Instance Method Summary
collapse
Constructor Details
#initialize(id: self.class.session_id, producer: self.class.producer) ⇒ KafkaSession
Returns a new instance of KafkaSession.
47
48
49
50
|
# File 'lib/kafka_session.rb', line 47
def initialize(id: self.class.session_id, producer: self.class.producer)
@id = id
@producer = producer
end
|
Class Attribute Details
.clock ⇒ Object
36
37
38
|
# File 'lib/kafka_session.rb', line 36
def self.clock
@clock ||= proc { Time.now.utc }
end
|
.producer ⇒ Object
24
25
26
|
# File 'lib/kafka_session.rb', line 24
def self.producer
@producer ||= MockProducer.new
end
|
Instance Attribute Details
#id ⇒ Object
Returns the value of attribute id.
44
45
46
|
# File 'lib/kafka_session.rb', line 44
def id
@id
end
|
#producer ⇒ Object
Returns the value of attribute producer.
45
46
47
|
# File 'lib/kafka_session.rb', line 45
def producer
@producer
end
|
Class Method Details
15
16
17
18
19
20
21
22
|
# File 'lib/kafka_session.rb', line 15
def self.configure(options = {})
producer_options = {
name: options.fetch(:name),
brokers: options.fetch(:brokers)
}
@producer = Producer.new(producer_options)
end
|
.now ⇒ Object
40
41
42
|
# File 'lib/kafka_session.rb', line 40
def self.now
clock.call
end
|
.session_id ⇒ Object
28
29
30
|
# File 'lib/kafka_session.rb', line 28
def self.session_id
Thread.current[:kafka_session_session_id] ||= SecureRandom.uuid
end
|
.session_id=(session_id) ⇒ Object
32
33
34
|
# File 'lib/kafka_session.rb', line 32
def self.session_id=(session_id)
Thread.current[:kafka_session_session_id] = session_id
end
|
Instance Method Details
#publish(topic, *message_values) ⇒ Object
52
53
54
55
56
57
58
|
# File 'lib/kafka_session.rb', line 52
def publish(topic, *message_values)
messages = message_values.map do |message_value|
Message.new(message_value, session_id: id).to_json
end
producer.publish(topic: topic, messages: messages)
end
|