| 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 (c *CreateTopicsRequest) requiredVersion() KafkaVersion { |
| switch c.Version { |
| case 2: |
| return V1_0_0_0 |
| case 1: |
| return V0_11_0_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 |
| } |