create_topics_request.go

package sarama

import (
	"time"
)

type CreateTopicsRequest struct {
	Version int16

	TopicDetails map[string]*TopicDetail
	Timeout      time.Duration
	ValidateOnly bool
}

func (c *CreateTopicsRequest) encode(pe packetEncoder) error {
	if err := pe.putArrayLength(len(c.TopicDetails)); err != nil {
		return err
	}
	for topic, detail := range c.TopicDetails {
		if err := pe.putString(topic); err != nil {
			return err
		}
		if err := detail.encode(pe); err != nil {
			return err
		}
	}

	pe.putInt32(int32(c.Timeout / time.Millisecond))

	if c.Version >= 1 {
		pe.putBool(c.ValidateOnly)
	}

	return nil
}

func (c *CreateTopicsRequest) decode(pd packetDecoder, version int16) (err error) {
	n, err := pd.getArrayLength()
	if err != nil {
		return err
	}

	c.TopicDetails = make(map[string]*TopicDetail, n)

	for i := 0; i < n; i++ {
		topic, err := pd.getString()
		if err != nil {
			return err
		}
		c.TopicDetails[topic] = new(TopicDetail)
		if err = c.TopicDetails[topic].decode(pd, version); err != nil {
			return err
		}
	}

	timeout, err := pd.getInt32()
	if err != nil {
		return err
	}
	c.Timeout = time.Duration(timeout) * time.Millisecond

	if version >= 1 {
		c.ValidateOnly, err = pd.getBool()
		if err != nil {
			return err
		}

		c.Version = version
	}

	return nil
}

func (c *CreateTopicsRequest) key() int16 {
	return 19
}

func (c *CreateTopicsRequest) version() int16 {
	return c.Version
}

func (r *CreateTopicsRequest) headerVersion() int16 {
	return 1
}

func (c *CreateTopicsRequest) isValidVersion() bool {
	return c.Version >= 0 && c.Version <= 3
}

func (c *CreateTopicsRequest) requiredVersion() KafkaVersion {
	switch c.Version {
	case 3:
		return V2_0_0_0
	case 2:
		return V0_11_0_0
	case 1:
		return V0_10_2_0
	default:
		return V0_10_1_0
	}
}

type TopicDetail struct {
	NumPartitions     int32
	ReplicationFactor int16
	ReplicaAssignment map[int32][]int32
	ConfigEntries     map[string]*string
}

func (t *TopicDetail) encode(pe packetEncoder) error {
	pe.putInt32(t.NumPartitions)
	pe.putInt16(t.ReplicationFactor)

	if err := pe.putArrayLength(len(t.ReplicaAssignment)); err != nil {
		return err
	}
	for partition, assignment := range t.ReplicaAssignment {
		pe.putInt32(partition)
		if err := pe.putInt32Array(assignment); err != nil {
			return err
		}
	}

	if err := pe.putArrayLength(len(t.ConfigEntries)); err != nil {
		return err
	}
	for configKey, configValue := range t.ConfigEntries {
		if err := pe.putString(configKey); err != nil {
			return err
		}
		if err := pe.putNullableString(configValue); err != nil {
			return err
		}
	}

	return nil
}

func (t *TopicDetail) decode(pd packetDecoder, version int16) (err error) {
	if t.NumPartitions, err = pd.getInt32(); err != nil {
		return err
	}
	if t.ReplicationFactor, err = pd.getInt16(); err != nil {
		return err
	}

	n, err := pd.getArrayLength()
	if err != nil {
		return err
	}

	if n > 0 {
		t.ReplicaAssignment = make(map[int32][]int32, n)
		for i := 0; i < n; i++ {
			replica, err := pd.getInt32()
			if err != nil {
				return err
			}
			if t.ReplicaAssignment[replica], err = pd.getInt32Array(); err != nil {
				return err
			}
		}
	}

	n, err = pd.getArrayLength()
	if err != nil {
		return err
	}

	if n > 0 {
		t.ConfigEntries = make(map[string]*string, n)
		for i := 0; i < n; i++ {
			configKey, err := pd.getString()
			if err != nil {
				return err
			}
			if t.ConfigEntries[configKey], err = pd.getNullableString(); err != nil {
				return err
			}
		}
	}

	return nil
}