cubefs

1
package sarama
2

3
type MetadataRequest struct {
4
	Version                int16
5
	Topics                 []string
6
	AllowAutoTopicCreation bool
7
}
8

9
func (r *MetadataRequest) encode(pe packetEncoder) error {
10
	if r.Version < 0 || r.Version > 5 {
11
		return PacketEncodingError{"invalid or unsupported MetadataRequest version field"}
12
	}
13
	if r.Version == 0 || len(r.Topics) > 0 {
14
		err := pe.putArrayLength(len(r.Topics))
15
		if err != nil {
16
			return err
17
		}
18

19
		for i := range r.Topics {
20
			err = pe.putString(r.Topics[i])
21
			if err != nil {
22
				return err
23
			}
24
		}
25
	} else {
26
		pe.putInt32(-1)
27
	}
28
	if r.Version > 3 {
29
		pe.putBool(r.AllowAutoTopicCreation)
30
	}
31
	return nil
32
}
33

34
func (r *MetadataRequest) decode(pd packetDecoder, version int16) error {
35
	r.Version = version
36
	size, err := pd.getInt32()
37
	if err != nil {
38
		return err
39
	}
40
	if size > 0 {
41
		r.Topics = make([]string, size)
42
		for i := range r.Topics {
43
			topic, err := pd.getString()
44
			if err != nil {
45
				return err
46
			}
47
			r.Topics[i] = topic
48
		}
49
	}
50
	if r.Version > 3 {
51
		autoCreation, err := pd.getBool()
52
		if err != nil {
53
			return err
54
		}
55
		r.AllowAutoTopicCreation = autoCreation
56
	}
57
	return nil
58
}
59

60
func (r *MetadataRequest) key() int16 {
61
	return 3
62
}
63

64
func (r *MetadataRequest) version() int16 {
65
	return r.Version
66
}
67

68
func (r *MetadataRequest) headerVersion() int16 {
69
	return 1
70
}
71

72
func (r *MetadataRequest) requiredVersion() KafkaVersion {
73
	switch r.Version {
74
	case 1:
75
		return V0_10_0_0
76
	case 2:
77
		return V0_10_1_0
78
	case 3, 4:
79
		return V0_11_0_0
80
	case 5:
81
		return V1_0_0_0
82
	default:
83
		return MinVersion
84
	}
85
}
86