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