Class: Kafka::Protocol::MemberAssignment

8
9
10
11
12

# File 'lib/kafka/protocol/member_assignment.rb', line 8
def initialize(version: 0, topics: {}, user_data: nil)
 @version = version
 @topics = topics
 @user_data = user_data
end

6
7
8

# File 'lib/kafka/protocol/member_assignment.rb', line 6
def topics
 @topics
end

33
34
35
36
37
38
39

# File 'lib/kafka/protocol/member_assignment.rb', line 33
def self.decode(decoder)
 new(
 version: decoder.int16,
 topics: Hash[decoder.array { [decoder.string, decoder.array { decoder.int32 }] }],
 user_data: decoder.bytes,
 )
end

14
15
16
17

# File 'lib/kafka/protocol/member_assignment.rb', line 14
def assign(topic, partitions)
 @topics[topic] ||= []
 @topics[topic].concat(partitions)
end

19
20
21
22
23
24
25
26
27
28
29
30
31

# File 'lib/kafka/protocol/member_assignment.rb', line 19
def encode(encoder)
 encoder.write_int16(@version)
 encoder.write_array(@topics) do |topic, partitions|
 encoder.write_string(topic)
 encoder.write_array(partitions) do |partition|
 encoder.write_int32(partition)
 end
 end
 encoder.write_bytes(@user_data)
end