sticky_assignor_user_data.go

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
}