Class: Kafka::MessageBuffer
- Inherits:
-
Object
- Object
- Kafka::MessageBuffer
- Includes:
- Enumerable
- Defined in:
- lib/kafka/message_buffer.rb
Overview
Buffers messages for specific topics/partitions.
Instance Attribute Summary collapse
-
#bytesize ⇒ Object
readonly
Returns the value of attribute bytesize.
-
#size ⇒ Object
readonly
Returns the value of attribute size.
Instance Method Summary collapse
-
#clear ⇒ nil
Clears messages across all topics and partitions.
-
#clear_messages(topic:, partition:) ⇒ nil
Clears buffered messages for the given topic and partition.
- #concat(messages, topic:, partition:) ⇒ Object
- #each ⇒ Object
- #empty? ⇒ Boolean
-
#initialize ⇒ MessageBuffer
constructor
A new instance of MessageBuffer.
- #messages_for(topic:, partition:) ⇒ Object
- #to_h ⇒ Object
- #write(value:, key:, topic:, partition:, create_time: Time.now) ⇒ Object
Constructor Details
#initialize ⇒ MessageBuffer
Returns a new instance of MessageBuffer.
11 12 13 14 15 |
# File 'lib/kafka/message_buffer.rb', line 11 def initialize @buffer = {} @size = 0 @bytesize = 0 end |
Instance Attribute Details
#bytesize ⇒ Object (readonly)
Returns the value of attribute bytesize.
9 10 11 |
# File 'lib/kafka/message_buffer.rb', line 9 def bytesize @bytesize end |
#size ⇒ Object (readonly)
Returns the value of attribute size.
9 10 11 |
# File 'lib/kafka/message_buffer.rb', line 9 def size @size end |
Instance Method Details
#clear ⇒ nil
Clears messages across all topics and partitions.
72 73 74 75 76 |
# File 'lib/kafka/message_buffer.rb', line 72 def clear @buffer = {} @size = 0 @bytesize = 0 end |
#clear_messages(topic:, partition:) ⇒ nil
Clears buffered messages for the given topic and partition.
55 56 57 58 59 60 61 62 63 |
# File 'lib/kafka/message_buffer.rb', line 55 def (topic:, partition:) return unless @buffer.key?(topic) && @buffer[topic].key?(partition) @size -= @buffer[topic][partition].count @bytesize -= @buffer[topic][partition].map(&:bytesize).reduce(0, :+) @buffer[topic].delete(partition) @buffer.delete(topic) if @buffer[topic].empty? end |
#concat(messages, topic:, partition:) ⇒ Object
26 27 28 29 30 31 |
# File 'lib/kafka/message_buffer.rb', line 26 def concat(, topic:, partition:) buffer_for(topic, partition).concat() @size += .count @bytesize += .map(&:bytesize).reduce(0, :+) end |
#each ⇒ Object
41 42 43 44 45 46 47 |
# File 'lib/kafka/message_buffer.rb', line 41 def each @buffer.each do |topic, | .each do |partition, | yield topic, partition, end end end |
#empty? ⇒ Boolean
37 38 39 |
# File 'lib/kafka/message_buffer.rb', line 37 def empty? @buffer.empty? end |
#messages_for(topic:, partition:) ⇒ Object
65 66 67 |
# File 'lib/kafka/message_buffer.rb', line 65 def (topic:, partition:) buffer_for(topic, partition) end |
#to_h ⇒ Object
33 34 35 |
# File 'lib/kafka/message_buffer.rb', line 33 def to_h @buffer end |
#write(value:, key:, topic:, partition:, create_time: Time.now) ⇒ Object
17 18 19 20 21 22 23 24 |
# File 'lib/kafka/message_buffer.rb', line 17 def write(value:, key:, topic:, partition:, create_time: Time.now) = Protocol::Message.new(key: key, value: value, create_time: create_time) buffer_for(topic, partition) << @size += 1 @bytesize += .bytesize end |