| package sarama |
| |
| type topicPartitionAssignment struct { |
| Topic string |
| Partition int32 |
| } |
| |
| type StickyAssignorUserData interface { |
| partitions() []topicPartitionAssignment |
| hasGeneration() bool |
| generation() int |
| } |
| |
| //StickyAssignorUserDataV0 holds topic partition information for an assignment |
| type StickyAssignorUserDataV0 struct { |
| Topics map[string][]int32 |
| |
| topicPartitions []topicPartitionAssignment |
| } |
| |
| func (m *StickyAssignorUserDataV0) encode(pe packetEncoder) error { |
| if err := pe.putArrayLength(len(m.Topics)); err != nil { |
| return err |
| } |
| |
| for topic, partitions := range m.Topics { |
| if err := pe.putString(topic); err != nil { |
| return err |
| } |
| if err := pe.putInt32Array(partitions); err != nil { |
| return err |
| } |
| } |
| return nil |
| } |
| |
| func (m *StickyAssignorUserDataV0) decode(pd packetDecoder) (err error) { |
| var topicLen int |
| if topicLen, err = pd.getArrayLength(); err != nil { |
| return |
| } |
| |
| m.Topics = make(map[string][]int32, topicLen) |
| for i := 0; i < topicLen; i++ { |
| var topic string |
| if topic, err = pd.getString(); err != nil { |
| return |
| } |
| if m.Topics[topic], err = pd.getInt32Array(); err != nil { |
| return |
| } |
| } |
| m.topicPartitions = populateTopicPartitions(m.Topics) |
| return nil |
| } |
| |
| func (m *StickyAssignorUserDataV0) partitions() []topicPartitionAssignment { return m.topicPartitions } |
| func (m *StickyAssignorUserDataV0) hasGeneration() bool { return false } |
| func (m *StickyAssignorUserDataV0) generation() int { return defaultGeneration } |
| |
| //StickyAssignorUserDataV1 holds topic partition information for an assignment |
| type StickyAssignorUserDataV1 struct { |
| Topics map[string][]int32 |
| Generation int32 |
| |
| topicPartitions []topicPartitionAssignment |
| } |
| |
| func (m *StickyAssignorUserDataV1) encode(pe packetEncoder) error { |
| if err := pe.putArrayLength(len(m.Topics)); err != nil { |
| return err |
| } |
| |
| for topic, partitions := range m.Topics { |
| if err := pe.putString(topic); err != nil { |
| return err |
| } |
| if err := pe.putInt32Array(partitions); err != nil { |
| return err |
| } |
| } |
| |
| pe.putInt32(m.Generation) |
| return nil |
| } |
| |
| func (m *StickyAssignorUserDataV1) decode(pd packetDecoder) (err error) { |
| var topicLen int |
| if topicLen, err = pd.getArrayLength(); err != nil { |
| return |
| } |
| |
| m.Topics = make(map[string][]int32, topicLen) |
| for i := 0; i < topicLen; i++ { |
| var topic string |
| if topic, err = pd.getString(); err != nil { |
| return |
| } |
| if m.Topics[topic], err = pd.getInt32Array(); err != nil { |
| return |
| } |
| } |
| |
| m.Generation, err = pd.getInt32() |
| if err != nil { |
| return err |
| } |
| m.topicPartitions = populateTopicPartitions(m.Topics) |
| return nil |
| } |
| |
| func (m *StickyAssignorUserDataV1) partitions() []topicPartitionAssignment { return m.topicPartitions } |
| func (m *StickyAssignorUserDataV1) hasGeneration() bool { return true } |
| func (m *StickyAssignorUserDataV1) generation() int { return int(m.Generation) } |
| |
| func populateTopicPartitions(topics map[string][]int32) []topicPartitionAssignment { |
| topicPartitions := make([]topicPartitionAssignment, 0) |
| for topic, partitions := range topics { |
| for _, partition := range partitions { |
| topicPartitions = append(topicPartitions, topicPartitionAssignment{Topic: topic, Partition: partition}) |
| } |
| } |
| return topicPartitions |
| } |