class Kafka::Protocol::MessageSet
def self.decode(decoder)
def self.decode(decoder) fetched_messages = [] until decoder.eof? begin message = Message.decode(decoder) if message.compressed? fetched_messages.concat(message.decompress) else fetched_messages << message end rescue EOFError if fetched_messages.empty? # If the first message in the set is truncated, it's likely because the # message is larger than the maximum size that we have asked for. raise MessageTooLargeToRead else # We tried to decode a partial message at the end of the set; just skip it. end end end new(messages: fetched_messages) end
def ==(other)
def ==(other) messages == other.messages end
def encode(encoder)
def encode(encoder) # Messages in a message set are *not* encoded as an array. Rather, # they are written in sequence. @messages.each do |message| message.encode(encoder) end end
def initialize(messages: [])
def initialize(messages: []) @messages = messages end
def size
def size @messages.size end