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.
- #message_count_for_partition(topic:, partition:) ⇒ Object
- #to_h ⇒ Object
- #write(value:, key:, topic:, partition:) ⇒ 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.
70 71 72 73 |
# File 'lib/kafka/message_buffer.rb', line 70 def clear @buffer = {} @size = 0 end |
#clear_messages(topic:, partition:) ⇒ nil
Clears buffered messages for the given topic and partition.
55 56 57 58 59 60 61 |
# File 'lib/kafka/message_buffer.rb', line 55 def (topic:, partition:) @size -= @buffer[topic][partition].count @bytesize -= @buffer[topic][partition].map(&:bytesize).reduce(:+) @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(:+) 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 |
#message_count_for_partition(topic:, partition:) ⇒ Object
63 64 65 |
# File 'lib/kafka/message_buffer.rb', line 63 def (topic:, partition:) buffer_for(topic, partition).count end |
#to_h ⇒ Object
33 34 35 |
# File 'lib/kafka/message_buffer.rb', line 33 def to_h @buffer end |