1use bytes::{Buf, BufMut};
4
5use crate::{
6 Decode, Encode, ProtocolError, UnknownTaggedFields,
7 primitives::{
8 fixed::{get_bool, get_i8, get_i16, get_i32, put_bool, put_i8, put_i16, put_i32},
9 string_bytes::{
10 compact_nullable_string_len, compact_string_len, get_compact_nullable_string_owned,
11 get_compact_string_owned, get_nullable_string_owned, get_string_owned,
12 nullable_string_len, put_compact_nullable_string, put_compact_string,
13 put_nullable_string, put_string, string_len,
14 },
15 },
16 tagged_fields::{WriteTaggedFields, read_tagged_fields, tagged_fields_len},
17};
18pub const API_KEY: i16 = 60;
19pub const MIN_VERSION: i16 = 0;
20pub const MAX_VERSION: i16 = 2;
21pub const FLEXIBLE_MIN: i16 = 0;
22#[inline]
23#[must_use]
24pub fn is_flexible(version: i16) -> bool {
25 version >= FLEXIBLE_MIN
26}
27#[derive(Debug, Clone, PartialEq, Eq)]
28pub struct DescribeClusterResponse {
29 pub throttle_time_ms: i32,
30 pub error_code: i16,
31 pub error_message: Option<String>,
32 pub endpoint_type: i8,
33 pub cluster_id: String,
34 pub controller_id: i32,
35 pub brokers: Vec<DescribeClusterBroker>,
36 pub cluster_authorized_operations: i32,
37 pub unknown_tagged_fields: UnknownTaggedFields,
38}
39impl Default for DescribeClusterResponse {
40 fn default() -> Self {
41 Self {
42 throttle_time_ms: 0i32,
43 error_code: 0i16,
44 error_message: None,
45 endpoint_type: 1i8,
46 cluster_id: String::new(),
47 controller_id: -1i32,
48 brokers: Vec::new(),
49 cluster_authorized_operations: -2_147_483_648i32,
50 unknown_tagged_fields: UnknownTaggedFields::default(),
51 }
52 }
53}
54impl DescribeClusterResponse {
55 fn encode_field_0<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
56 if version >= 0 {
57 put_i32(buf, self.throttle_time_ms);
58 }
59 }
60 fn encode_field_1<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
61 if version >= 0 {
62 put_i16(buf, self.error_code);
63 }
64 }
65 fn encode_field_2<B: BufMut>(&self, buf: &mut B, version: i16, flex: bool) {
66 if version >= 0 {
67 if flex {
68 let () = put_compact_nullable_string(buf, self.error_message.as_deref());
69 } else {
70 let () = put_nullable_string(buf, self.error_message.as_deref());
71 }
72 }
73 }
74 fn encode_field_3<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
75 if version >= 1 {
76 put_i8(buf, self.endpoint_type);
77 }
78 }
79 fn encode_field_4<B: BufMut>(&self, buf: &mut B, version: i16, flex: bool) {
80 if version >= 0 {
81 if flex {
82 let () = put_compact_string(buf, &self.cluster_id);
83 } else {
84 let () = put_string(buf, &self.cluster_id);
85 }
86 }
87 }
88 fn encode_field_5<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
89 if version >= 0 {
90 put_i32(buf, self.controller_id);
91 }
92 }
93 fn encode_field_6<B: BufMut>(
94 &self,
95 buf: &mut B,
96 version: i16,
97 flex: bool,
98 ) -> Result<(), ProtocolError> {
99 if version >= 0 {
100 {
101 crate::primitives::array::put_array_len(buf, (self.brokers).len(), flex);
102 for it in &self.brokers {
103 it.encode(buf, version)?;
104 }
105 };
106 }
107 Ok(())
108 }
109 fn encode_field_7<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
110 if version >= 0 {
111 put_i32(buf, self.cluster_authorized_operations);
112 }
113 }
114 fn encode_tagged_fields<B: BufMut>(&self, buf: &mut B, _version: i16, flex: bool) {
115 if flex {
116 let tagged = WriteTaggedFields::new();
117 tagged.write(buf, &self.unknown_tagged_fields);
118 }
119 }
120 fn decode_field_0<B: Buf>(
121 out: &mut Self,
122 buf: &mut B,
123 version: i16,
124 _flex: bool,
125 ) -> Result<(), ProtocolError> {
126 if version >= 0 {
127 out.throttle_time_ms = get_i32(buf)?;
128 }
129 Ok(())
130 }
131 fn decode_field_1<B: Buf>(
132 out: &mut Self,
133 buf: &mut B,
134 version: i16,
135 _flex: bool,
136 ) -> Result<(), ProtocolError> {
137 if version >= 0 {
138 out.error_code = get_i16(buf)?;
139 }
140 Ok(())
141 }
142 fn decode_field_2<B: Buf>(
143 out: &mut Self,
144 buf: &mut B,
145 version: i16,
146 flex: bool,
147 ) -> Result<(), ProtocolError> {
148 if version >= 0 {
149 out.error_message = if flex {
150 get_compact_nullable_string_owned(buf)?
151 } else {
152 get_nullable_string_owned(buf)?
153 };
154 }
155 Ok(())
156 }
157 fn decode_field_3<B: Buf>(
158 out: &mut Self,
159 buf: &mut B,
160 version: i16,
161 _flex: bool,
162 ) -> Result<(), ProtocolError> {
163 if version >= 1 {
164 out.endpoint_type = get_i8(buf)?;
165 }
166 Ok(())
167 }
168 fn decode_field_4<B: Buf>(
169 out: &mut Self,
170 buf: &mut B,
171 version: i16,
172 flex: bool,
173 ) -> Result<(), ProtocolError> {
174 if version >= 0 {
175 out.cluster_id = if flex {
176 get_compact_string_owned(buf)?
177 } else {
178 get_string_owned(buf)?
179 };
180 }
181 Ok(())
182 }
183 fn decode_field_5<B: Buf>(
184 out: &mut Self,
185 buf: &mut B,
186 version: i16,
187 _flex: bool,
188 ) -> Result<(), ProtocolError> {
189 if version >= 0 {
190 out.controller_id = get_i32(buf)?;
191 }
192 Ok(())
193 }
194 fn decode_field_6<B: Buf>(
195 out: &mut Self,
196 buf: &mut B,
197 version: i16,
198 flex: bool,
199 ) -> Result<(), ProtocolError> {
200 if version >= 0 {
201 out.brokers = {
202 let n = crate::primitives::array::get_array_len(buf, flex)?;
203 let mut v = Vec::with_capacity(n);
204 for _ in 0..n {
205 v.push(DescribeClusterBroker::decode(buf, version)?);
206 }
207 v
208 };
209 }
210 Ok(())
211 }
212 fn decode_field_7<B: Buf>(
213 out: &mut Self,
214 buf: &mut B,
215 version: i16,
216 _flex: bool,
217 ) -> Result<(), ProtocolError> {
218 if version >= 0 {
219 out.cluster_authorized_operations = get_i32(buf)?;
220 }
221 Ok(())
222 }
223 fn decode_tagged_fields<B: Buf>(
224 out: &mut Self,
225 buf: &mut B,
226 _version: i16,
227 flex: bool,
228 ) -> Result<(), ProtocolError> {
229 if flex {
230 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
231 }
232 Ok(())
233 }
234}
235impl Encode for DescribeClusterResponse {
236 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
237 if !(MIN_VERSION..=MAX_VERSION).contains(&version) {
238 return Err(ProtocolError::UnsupportedVersion {
239 api_key: API_KEY,
240 version,
241 });
242 }
243 let flex = is_flexible(version);
244 self.encode_field_0(buf, version, flex);
245 self.encode_field_1(buf, version, flex);
246 self.encode_field_2(buf, version, flex);
247 self.encode_field_3(buf, version, flex);
248 self.encode_field_4(buf, version, flex);
249 self.encode_field_5(buf, version, flex);
250 self.encode_field_6(buf, version, flex)?;
251 self.encode_field_7(buf, version, flex);
252 self.encode_tagged_fields(buf, version, flex);
253 Ok(())
254 }
255 fn encoded_len(&self, version: i16) -> usize {
256 let flex = is_flexible(version);
257 let mut n: usize = 0;
258 if version >= 0 {
259 n += 4;
260 }
261 if version >= 0 {
262 n += 2;
263 }
264 if version >= 0 {
265 n += if flex {
266 compact_nullable_string_len(self.error_message.as_deref())
267 } else {
268 nullable_string_len(self.error_message.as_deref())
269 };
270 }
271 if version >= 1 {
272 n += 1;
273 }
274 if version >= 0 {
275 n += if flex {
276 compact_string_len(&self.cluster_id)
277 } else {
278 string_len(&self.cluster_id)
279 };
280 }
281 if version >= 0 {
282 n += 4;
283 }
284 if version >= 0 {
285 n += {
286 let prefix =
287 crate::primitives::array::array_len_prefix_len((self.brokers).len(), flex);
288 let body: usize = (self.brokers)
289 .iter()
290 .map(|it| it.encoded_len(version))
291 .sum();
292 prefix + body
293 };
294 }
295 if version >= 0 {
296 n += 4;
297 }
298 if flex {
299 let known_pairs: Vec<(u32, usize)> = Vec::new();
300 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
301 }
302 n
303 }
304}
305impl Decode<'_> for DescribeClusterResponse {
306 fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
307 if !(MIN_VERSION..=MAX_VERSION).contains(&version) {
308 return Err(ProtocolError::UnsupportedVersion {
309 api_key: API_KEY,
310 version,
311 });
312 }
313 let flex = is_flexible(version);
314 let mut out = Self::default();
315 Self::decode_field_0(&mut out, buf, version, flex)?;
316 Self::decode_field_1(&mut out, buf, version, flex)?;
317 Self::decode_field_2(&mut out, buf, version, flex)?;
318 Self::decode_field_3(&mut out, buf, version, flex)?;
319 Self::decode_field_4(&mut out, buf, version, flex)?;
320 Self::decode_field_5(&mut out, buf, version, flex)?;
321 Self::decode_field_6(&mut out, buf, version, flex)?;
322 Self::decode_field_7(&mut out, buf, version, flex)?;
323 Self::decode_tagged_fields(&mut out, buf, version, flex)?;
324 Ok(out)
325 }
326}
327#[cfg(test)]
328impl DescribeClusterResponse {
329 #[must_use]
330 pub fn populated(version: i16) -> Self {
331 let mut m = Self::default();
332 if version >= 0 {
333 m.throttle_time_ms = 1i32;
334 }
335 if version >= 0 {
336 m.error_code = 1i16;
337 }
338 if version >= 0 {
339 m.error_message = Some("x".to_string());
340 }
341 if version >= 1 {
342 m.endpoint_type = 1i8;
343 }
344 if version >= 0 {
345 m.cluster_id = "x".to_string();
346 }
347 if version >= 0 {
348 m.controller_id = 1i32;
349 }
350 if version >= 0 {
351 m.brokers = vec![DescribeClusterBroker::populated(version)];
352 }
353 if version >= 0 {
354 m.cluster_authorized_operations = 1i32;
355 }
356 m
357 }
358}
359#[derive(Debug, Clone, PartialEq, Eq, Default)]
360pub struct DescribeClusterBroker {
361 pub broker_id: i32,
362 pub host: String,
363 pub port: i32,
364 pub rack: Option<String>,
365 pub is_fenced: bool,
366 pub unknown_tagged_fields: UnknownTaggedFields,
367}
368impl Encode for DescribeClusterBroker {
369 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
370 let flex = version >= 0;
371 if version >= 0 {
372 put_i32(buf, self.broker_id);
373 }
374 if version >= 0 {
375 if flex {
376 let () = put_compact_string(buf, &self.host);
377 } else {
378 let () = put_string(buf, &self.host);
379 }
380 }
381 if version >= 0 {
382 put_i32(buf, self.port);
383 }
384 if version >= 0 {
385 if flex {
386 let () = put_compact_nullable_string(buf, self.rack.as_deref());
387 } else {
388 let () = put_nullable_string(buf, self.rack.as_deref());
389 }
390 }
391 if version >= 2 {
392 put_bool(buf, self.is_fenced);
393 }
394 if flex {
395 let tagged = WriteTaggedFields::new();
396 tagged.write(buf, &self.unknown_tagged_fields);
397 }
398 Ok(())
399 }
400 fn encoded_len(&self, version: i16) -> usize {
401 let flex = version >= 0;
402 let mut n: usize = 0;
403 if version >= 0 {
404 n += 4;
405 }
406 if version >= 0 {
407 n += if flex {
408 compact_string_len(&self.host)
409 } else {
410 string_len(&self.host)
411 };
412 }
413 if version >= 0 {
414 n += 4;
415 }
416 if version >= 0 {
417 n += if flex {
418 compact_nullable_string_len(self.rack.as_deref())
419 } else {
420 nullable_string_len(self.rack.as_deref())
421 };
422 }
423 if version >= 2 {
424 n += 1;
425 }
426 if flex {
427 let known_pairs: Vec<(u32, usize)> = Vec::new();
428 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
429 }
430 n
431 }
432}
433impl Decode<'_> for DescribeClusterBroker {
434 fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
435 let flex = version >= 0;
436 let mut out = Self::default();
437 if version >= 0 {
438 out.broker_id = get_i32(buf)?;
439 }
440 if version >= 0 {
441 out.host = if flex {
442 get_compact_string_owned(buf)?
443 } else {
444 get_string_owned(buf)?
445 };
446 }
447 if version >= 0 {
448 out.port = get_i32(buf)?;
449 }
450 if version >= 0 {
451 out.rack = if flex {
452 get_compact_nullable_string_owned(buf)?
453 } else {
454 get_nullable_string_owned(buf)?
455 };
456 }
457 if version >= 2 {
458 out.is_fenced = get_bool(buf)?;
459 }
460 if flex {
461 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
462 }
463 Ok(out)
464 }
465}
466#[cfg(test)]
467impl DescribeClusterBroker {
468 #[must_use]
469 pub fn populated(version: i16) -> Self {
470 let mut m = Self::default();
471 if version >= 0 {
472 m.broker_id = 1i32;
473 }
474 if version >= 0 {
475 m.host = "x".to_string();
476 }
477 if version >= 0 {
478 m.port = 1i32;
479 }
480 if version >= 0 {
481 m.rack = Some("x".to_string());
482 }
483 if version >= 2 {
484 m.is_fenced = true;
485 }
486 m
487 }
488}
489#[must_use]
492#[allow(unused_comparisons)]
493pub fn default_json(version: i16) -> ::serde_json::Value {
494 let mut obj = ::serde_json::Map::new();
495 obj.insert("throttleTimeMs".to_string(), ::serde_json::json!(0));
496 obj.insert("errorCode".to_string(), ::serde_json::json!(0));
497 obj.insert("errorMessage".to_string(), ::serde_json::Value::Null);
498 if version >= 1 {
499 obj.insert("endpointType".to_string(), ::serde_json::json!(1));
500 }
501 obj.insert(
502 "clusterId".to_string(),
503 ::serde_json::Value::String(String::new()),
504 );
505 obj.insert("controllerId".to_string(), ::serde_json::json!(-1));
506 obj.insert("brokers".to_string(), ::serde_json::Value::Array(vec![]));
507 obj.insert(
508 "clusterAuthorizedOperations".to_string(),
509 ::serde_json::json!(-2_147_483_648),
510 );
511 ::serde_json::Value::Object(obj)
512}