cubefs

1
package sarama
2

3
type DescribeGroupsResponse struct {
4
	Groups []*GroupDescription
5
}
6

7
func (r *DescribeGroupsResponse) encode(pe packetEncoder) error {
8
	if err := pe.putArrayLength(len(r.Groups)); err != nil {
9
		return err
10
	}
11

12
	for _, groupDescription := range r.Groups {
13
		if err := groupDescription.encode(pe); err != nil {
14
			return err
15
		}
16
	}
17

18
	return nil
19
}
20

21
func (r *DescribeGroupsResponse) decode(pd packetDecoder, version int16) (err error) {
22
	n, err := pd.getArrayLength()
23
	if err != nil {
24
		return err
25
	}
26

27
	r.Groups = make([]*GroupDescription, n)
28
	for i := 0; i < n; i++ {
29
		r.Groups[i] = new(GroupDescription)
30
		if err := r.Groups[i].decode(pd); err != nil {
31
			return err
32
		}
33
	}
34

35
	return nil
36
}
37

38
func (r *DescribeGroupsResponse) key() int16 {
39
	return 15
40
}
41

42
func (r *DescribeGroupsResponse) version() int16 {
43
	return 0
44
}
45

46
func (r *DescribeGroupsResponse) headerVersion() int16 {
47
	return 0
48
}
49

50
func (r *DescribeGroupsResponse) requiredVersion() KafkaVersion {
51
	return V0_9_0_0
52
}
53

54
type GroupDescription struct {
55
	Err          KError
56
	GroupId      string
57
	State        string
58
	ProtocolType string
59
	Protocol     string
60
	Members      map[string]*GroupMemberDescription
61
}
62

63
func (gd *GroupDescription) encode(pe packetEncoder) error {
64
	pe.putInt16(int16(gd.Err))
65

66
	if err := pe.putString(gd.GroupId); err != nil {
67
		return err
68
	}
69
	if err := pe.putString(gd.State); err != nil {
70
		return err
71
	}
72
	if err := pe.putString(gd.ProtocolType); err != nil {
73
		return err
74
	}
75
	if err := pe.putString(gd.Protocol); err != nil {
76
		return err
77
	}
78

79
	if err := pe.putArrayLength(len(gd.Members)); err != nil {
80
		return err
81
	}
82

83
	for memberId, groupMemberDescription := range gd.Members {
84
		if err := pe.putString(memberId); err != nil {
85
			return err
86
		}
87
		if err := groupMemberDescription.encode(pe); err != nil {
88
			return err
89
		}
90
	}
91

92
	return nil
93
}
94

95
func (gd *GroupDescription) decode(pd packetDecoder) (err error) {
96
	kerr, err := pd.getInt16()
97
	if err != nil {
98
		return err
99
	}
100

101
	gd.Err = KError(kerr)
102

103
	if gd.GroupId, err = pd.getString(); err != nil {
104
		return
105
	}
106
	if gd.State, err = pd.getString(); err != nil {
107
		return
108
	}
109
	if gd.ProtocolType, err = pd.getString(); err != nil {
110
		return
111
	}
112
	if gd.Protocol, err = pd.getString(); err != nil {
113
		return
114
	}
115

116
	n, err := pd.getArrayLength()
117
	if err != nil {
118
		return err
119
	}
120
	if n == 0 {
121
		return nil
122
	}
123

124
	gd.Members = make(map[string]*GroupMemberDescription)
125
	for i := 0; i < n; i++ {
126
		memberId, err := pd.getString()
127
		if err != nil {
128
			return err
129
		}
130

131
		gd.Members[memberId] = new(GroupMemberDescription)
132
		if err := gd.Members[memberId].decode(pd); err != nil {
133
			return err
134
		}
135
	}
136

137
	return nil
138
}
139

140
type GroupMemberDescription struct {
141
	ClientId         string
142
	ClientHost       string
143
	MemberMetadata   []byte
144
	MemberAssignment []byte
145
}
146

147
func (gmd *GroupMemberDescription) encode(pe packetEncoder) error {
148
	if err := pe.putString(gmd.ClientId); err != nil {
149
		return err
150
	}
151
	if err := pe.putString(gmd.ClientHost); err != nil {
152
		return err
153
	}
154
	if err := pe.putBytes(gmd.MemberMetadata); err != nil {
155
		return err
156
	}
157
	if err := pe.putBytes(gmd.MemberAssignment); err != nil {
158
		return err
159
	}
160

161
	return nil
162
}
163

164
func (gmd *GroupMemberDescription) decode(pd packetDecoder) (err error) {
165
	if gmd.ClientId, err = pd.getString(); err != nil {
166
		return
167
	}
168
	if gmd.ClientHost, err = pd.getString(); err != nil {
169
		return
170
	}
171
	if gmd.MemberMetadata, err = pd.getBytes(); err != nil {
172
		return
173
	}
174
	if gmd.MemberAssignment, err = pd.getBytes(); err != nil {
175
		return
176
	}
177

178
	return nil
179
}
180

181
func (gmd *GroupMemberDescription) GetMemberAssignment() (*ConsumerGroupMemberAssignment, error) {
182
	if len(gmd.MemberAssignment) == 0 {
183
		return nil, nil
184
	}
185
	assignment := new(ConsumerGroupMemberAssignment)
186
	err := decode(gmd.MemberAssignment, assignment)
187
	return assignment, err
188
}
189

190
func (gmd *GroupMemberDescription) GetMemberMetadata() (*ConsumerGroupMemberMetadata, error) {
191
	if len(gmd.MemberMetadata) == 0 {
192
		return nil, nil
193
	}
194
	metadata := new(ConsumerGroupMemberMetadata)
195
	err := decode(gmd.MemberMetadata, metadata)
196
	return metadata, err
197
}
198