Sha256: 939cfd2870b7eb7670627bdcfc5909e256999bf4c6194fcf3350a8e28b718d74
Contents?: true
Size: 1.05 KB
Versions: 39
Compression:
Stored size: 1.05 KB
Contents
module Kafka
module Protocol
class MessageSet
attr_reader :messages
def initialize(messages: [])
@messages = messages
end
def size
@messages.size
end
def ==(other)
messages == other.messages
end
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 self.decode(decoder)
fetched_messages = []
until decoder.eof?
begin
message = Message.decode(decoder)
if message.compressed?
wrapped_message_set = message.decompress
fetched_messages.concat(wrapped_message_set.messages)
else
fetched_messages << message
end
rescue EOFError
# We tried to decode a partial message; just skip it.
end
end
new(messages: fetched_messages)
end
end
end
end
Version data entries
39 entries across 39 versions & 1 rubygems