Sha256: b40fabbf0d5290191962f958320f71b3776726f8a4ba5298522652df9b6bf6ea
Contents?: true
Size: 1.9 KB
Versions: 17
Compression:
Stored size: 1.9 KB
Contents
require "kafka/protocol/message"
module Kafka
# Buffers messages for specific topics/partitions.
class MessageBuffer
include Enumerable
attr_reader :size, :bytesize
def initialize
@buffer = {}
@size = 0
@bytesize = 0
end
def write(value:, key:, topic:, partition:, create_time: Time.now)
message = Protocol::Message.new(key: key, value: value, create_time: create_time)
buffer_for(topic, partition) << message
@size += 1
@bytesize += message.bytesize
end
def concat(messages, topic:, partition:)
buffer_for(topic, partition).concat(messages)
@size += messages.count
@bytesize += messages.map(&:bytesize).reduce(0, :+)
end
def to_h
@buffer
end
def empty?
@buffer.empty?
end
def each
@buffer.each do |topic, messages_for_topic|
messages_for_topic.each do |partition, messages_for_partition|
yield topic, partition, messages_for_partition
end
end
end
# Clears buffered messages for the given topic and partition.
#
# @param topic [String] the name of the topic.
# @param partition [Integer] the partition id.
#
# @return [nil]
def clear_messages(topic:, partition:)
return unless @buffer.key?(topic) && @buffer[topic].key?(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
def messages_for(topic:, partition:)
buffer_for(topic, partition)
end
# Clears messages across all topics and partitions.
#
# @return [nil]
def clear
@buffer = {}
@size = 0
@bytesize = 0
end
private
def buffer_for(topic, partition)
@buffer[topic] ||= {}
@buffer[topic][partition] ||= []
end
end
end
Version data entries
17 entries across 17 versions & 1 rubygems