1use bytes::{Buf, BufMut};
4
5use crate::{
6 Decode, Encode, ProtocolError, UnknownTaggedFields,
7 primitives::{
8 fixed::{get_i8, get_i16, get_i32, 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 = 69;
19pub const MIN_VERSION: i16 = 0;
20pub const MAX_VERSION: i16 = 1;
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, Default)]
28pub struct ConsumerGroupDescribeResponse {
29 pub throttle_time_ms: i32,
30 pub groups: Vec<DescribedGroup>,
31 pub unknown_tagged_fields: UnknownTaggedFields,
32}
33impl Encode for ConsumerGroupDescribeResponse {
34 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
35 if !(MIN_VERSION..=MAX_VERSION).contains(&version) {
36 return Err(ProtocolError::UnsupportedVersion {
37 api_key: API_KEY,
38 version,
39 });
40 }
41 let flex = is_flexible(version);
42 if version >= 0 {
43 put_i32(buf, self.throttle_time_ms);
44 }
45 if version >= 0 {
46 {
47 crate::primitives::array::put_array_len(buf, (self.groups).len(), flex);
48 for it in &self.groups {
49 it.encode(buf, version)?;
50 }
51 };
52 }
53 if flex {
54 let tagged = WriteTaggedFields::new();
55 tagged.write(buf, &self.unknown_tagged_fields);
56 }
57 Ok(())
58 }
59 fn encoded_len(&self, version: i16) -> usize {
60 let flex = is_flexible(version);
61 let mut n: usize = 0;
62 if version >= 0 {
63 n += 4;
64 }
65 if version >= 0 {
66 n += {
67 let prefix =
68 crate::primitives::array::array_len_prefix_len((self.groups).len(), flex);
69 let body: usize = (self.groups).iter().map(|it| it.encoded_len(version)).sum();
70 prefix + body
71 };
72 }
73 if flex {
74 let known_pairs: Vec<(u32, usize)> = Vec::new();
75 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
76 }
77 n
78 }
79}
80impl Decode<'_> for ConsumerGroupDescribeResponse {
81 fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
82 if !(MIN_VERSION..=MAX_VERSION).contains(&version) {
83 return Err(ProtocolError::UnsupportedVersion {
84 api_key: API_KEY,
85 version,
86 });
87 }
88 let flex = is_flexible(version);
89 let mut out = Self::default();
90 if version >= 0 {
91 out.throttle_time_ms = get_i32(buf)?;
92 }
93 if version >= 0 {
94 out.groups = {
95 let n = crate::primitives::array::get_array_len(buf, flex)?;
96 let mut v = Vec::with_capacity(n);
97 for _ in 0..n {
98 v.push(DescribedGroup::decode(buf, version)?);
99 }
100 v
101 };
102 }
103 if flex {
104 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
105 }
106 Ok(out)
107 }
108}
109#[cfg(test)]
110impl ConsumerGroupDescribeResponse {
111 #[must_use]
112 pub fn populated(version: i16) -> Self {
113 let mut m = Self::default();
114 if version >= 0 {
115 m.throttle_time_ms = 1i32;
116 }
117 if version >= 0 {
118 m.groups = vec![DescribedGroup::populated(version)];
119 }
120 m
121 }
122}
123#[derive(Debug, Clone, PartialEq, Eq)]
124pub struct DescribedGroup {
125 pub error_code: i16,
126 pub error_message: Option<String>,
127 pub group_id: String,
128 pub group_state: String,
129 pub group_epoch: i32,
130 pub assignment_epoch: i32,
131 pub assignor_name: String,
132 pub members: Vec<Member>,
133 pub authorized_operations: i32,
134 pub unknown_tagged_fields: UnknownTaggedFields,
135}
136impl Default for DescribedGroup {
137 fn default() -> Self {
138 Self {
139 error_code: 0i16,
140 error_message: None,
141 group_id: String::new(),
142 group_state: String::new(),
143 group_epoch: 0i32,
144 assignment_epoch: 0i32,
145 assignor_name: String::new(),
146 members: Vec::new(),
147 authorized_operations: -2_147_483_648i32,
148 unknown_tagged_fields: UnknownTaggedFields::default(),
149 }
150 }
151}
152impl DescribedGroup {
153 fn encode_field_0<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
154 if version >= 0 {
155 put_i16(buf, self.error_code);
156 }
157 }
158 fn encode_field_1<B: BufMut>(&self, buf: &mut B, version: i16, flex: bool) {
159 if version >= 0 {
160 if flex {
161 let () = put_compact_nullable_string(buf, self.error_message.as_deref());
162 } else {
163 let () = put_nullable_string(buf, self.error_message.as_deref());
164 }
165 }
166 }
167 fn encode_field_2<B: BufMut>(&self, buf: &mut B, version: i16, flex: bool) {
168 if version >= 0 {
169 if flex {
170 let () = put_compact_string(buf, &self.group_id);
171 } else {
172 let () = put_string(buf, &self.group_id);
173 }
174 }
175 }
176 fn encode_field_3<B: BufMut>(&self, buf: &mut B, version: i16, flex: bool) {
177 if version >= 0 {
178 if flex {
179 let () = put_compact_string(buf, &self.group_state);
180 } else {
181 let () = put_string(buf, &self.group_state);
182 }
183 }
184 }
185 fn encode_field_4<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
186 if version >= 0 {
187 put_i32(buf, self.group_epoch);
188 }
189 }
190 fn encode_field_5<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
191 if version >= 0 {
192 put_i32(buf, self.assignment_epoch);
193 }
194 }
195 fn encode_field_6<B: BufMut>(&self, buf: &mut B, version: i16, flex: bool) {
196 if version >= 0 {
197 if flex {
198 let () = put_compact_string(buf, &self.assignor_name);
199 } else {
200 let () = put_string(buf, &self.assignor_name);
201 }
202 }
203 }
204 fn encode_field_7<B: BufMut>(
205 &self,
206 buf: &mut B,
207 version: i16,
208 flex: bool,
209 ) -> Result<(), ProtocolError> {
210 if version >= 0 {
211 {
212 crate::primitives::array::put_array_len(buf, (self.members).len(), flex);
213 for it in &self.members {
214 it.encode(buf, version)?;
215 }
216 };
217 }
218 Ok(())
219 }
220 fn encode_field_8<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
221 if version >= 0 {
222 put_i32(buf, self.authorized_operations);
223 }
224 }
225 fn encode_tagged_fields<B: BufMut>(&self, buf: &mut B, _version: i16, flex: bool) {
226 if flex {
227 let tagged = WriteTaggedFields::new();
228 tagged.write(buf, &self.unknown_tagged_fields);
229 }
230 }
231 fn decode_field_0<B: Buf>(
232 out: &mut Self,
233 buf: &mut B,
234 version: i16,
235 _flex: bool,
236 ) -> Result<(), ProtocolError> {
237 if version >= 0 {
238 out.error_code = get_i16(buf)?;
239 }
240 Ok(())
241 }
242 fn decode_field_1<B: Buf>(
243 out: &mut Self,
244 buf: &mut B,
245 version: i16,
246 flex: bool,
247 ) -> Result<(), ProtocolError> {
248 if version >= 0 {
249 out.error_message = if flex {
250 get_compact_nullable_string_owned(buf)?
251 } else {
252 get_nullable_string_owned(buf)?
253 };
254 }
255 Ok(())
256 }
257 fn decode_field_2<B: Buf>(
258 out: &mut Self,
259 buf: &mut B,
260 version: i16,
261 flex: bool,
262 ) -> Result<(), ProtocolError> {
263 if version >= 0 {
264 out.group_id = if flex {
265 get_compact_string_owned(buf)?
266 } else {
267 get_string_owned(buf)?
268 };
269 }
270 Ok(())
271 }
272 fn decode_field_3<B: Buf>(
273 out: &mut Self,
274 buf: &mut B,
275 version: i16,
276 flex: bool,
277 ) -> Result<(), ProtocolError> {
278 if version >= 0 {
279 out.group_state = if flex {
280 get_compact_string_owned(buf)?
281 } else {
282 get_string_owned(buf)?
283 };
284 }
285 Ok(())
286 }
287 fn decode_field_4<B: Buf>(
288 out: &mut Self,
289 buf: &mut B,
290 version: i16,
291 _flex: bool,
292 ) -> Result<(), ProtocolError> {
293 if version >= 0 {
294 out.group_epoch = get_i32(buf)?;
295 }
296 Ok(())
297 }
298 fn decode_field_5<B: Buf>(
299 out: &mut Self,
300 buf: &mut B,
301 version: i16,
302 _flex: bool,
303 ) -> Result<(), ProtocolError> {
304 if version >= 0 {
305 out.assignment_epoch = get_i32(buf)?;
306 }
307 Ok(())
308 }
309 fn decode_field_6<B: Buf>(
310 out: &mut Self,
311 buf: &mut B,
312 version: i16,
313 flex: bool,
314 ) -> Result<(), ProtocolError> {
315 if version >= 0 {
316 out.assignor_name = if flex {
317 get_compact_string_owned(buf)?
318 } else {
319 get_string_owned(buf)?
320 };
321 }
322 Ok(())
323 }
324 fn decode_field_7<B: Buf>(
325 out: &mut Self,
326 buf: &mut B,
327 version: i16,
328 flex: bool,
329 ) -> Result<(), ProtocolError> {
330 if version >= 0 {
331 out.members = {
332 let n = crate::primitives::array::get_array_len(buf, flex)?;
333 let mut v = Vec::with_capacity(n);
334 for _ in 0..n {
335 v.push(Member::decode(buf, version)?);
336 }
337 v
338 };
339 }
340 Ok(())
341 }
342 fn decode_field_8<B: Buf>(
343 out: &mut Self,
344 buf: &mut B,
345 version: i16,
346 _flex: bool,
347 ) -> Result<(), ProtocolError> {
348 if version >= 0 {
349 out.authorized_operations = get_i32(buf)?;
350 }
351 Ok(())
352 }
353 fn decode_tagged_fields<B: Buf>(
354 out: &mut Self,
355 buf: &mut B,
356 _version: i16,
357 flex: bool,
358 ) -> Result<(), ProtocolError> {
359 if flex {
360 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
361 }
362 Ok(())
363 }
364}
365impl Encode for DescribedGroup {
366 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
367 let flex = version >= 0;
368 self.encode_field_0(buf, version, flex);
369 self.encode_field_1(buf, version, flex);
370 self.encode_field_2(buf, version, flex);
371 self.encode_field_3(buf, version, flex);
372 self.encode_field_4(buf, version, flex);
373 self.encode_field_5(buf, version, flex);
374 self.encode_field_6(buf, version, flex);
375 self.encode_field_7(buf, version, flex)?;
376 self.encode_field_8(buf, version, flex);
377 self.encode_tagged_fields(buf, version, flex);
378 Ok(())
379 }
380 fn encoded_len(&self, version: i16) -> usize {
381 let flex = version >= 0;
382 let mut n: usize = 0;
383 if version >= 0 {
384 n += 2;
385 }
386 if version >= 0 {
387 n += if flex {
388 compact_nullable_string_len(self.error_message.as_deref())
389 } else {
390 nullable_string_len(self.error_message.as_deref())
391 };
392 }
393 if version >= 0 {
394 n += if flex {
395 compact_string_len(&self.group_id)
396 } else {
397 string_len(&self.group_id)
398 };
399 }
400 if version >= 0 {
401 n += if flex {
402 compact_string_len(&self.group_state)
403 } else {
404 string_len(&self.group_state)
405 };
406 }
407 if version >= 0 {
408 n += 4;
409 }
410 if version >= 0 {
411 n += 4;
412 }
413 if version >= 0 {
414 n += if flex {
415 compact_string_len(&self.assignor_name)
416 } else {
417 string_len(&self.assignor_name)
418 };
419 }
420 if version >= 0 {
421 n += {
422 let prefix =
423 crate::primitives::array::array_len_prefix_len((self.members).len(), flex);
424 let body: usize = (self.members)
425 .iter()
426 .map(|it| it.encoded_len(version))
427 .sum();
428 prefix + body
429 };
430 }
431 if version >= 0 {
432 n += 4;
433 }
434 if flex {
435 let known_pairs: Vec<(u32, usize)> = Vec::new();
436 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
437 }
438 n
439 }
440}
441impl Decode<'_> for DescribedGroup {
442 fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
443 let flex = version >= 0;
444 let mut out = Self::default();
445 Self::decode_field_0(&mut out, buf, version, flex)?;
446 Self::decode_field_1(&mut out, buf, version, flex)?;
447 Self::decode_field_2(&mut out, buf, version, flex)?;
448 Self::decode_field_3(&mut out, buf, version, flex)?;
449 Self::decode_field_4(&mut out, buf, version, flex)?;
450 Self::decode_field_5(&mut out, buf, version, flex)?;
451 Self::decode_field_6(&mut out, buf, version, flex)?;
452 Self::decode_field_7(&mut out, buf, version, flex)?;
453 Self::decode_field_8(&mut out, buf, version, flex)?;
454 Self::decode_tagged_fields(&mut out, buf, version, flex)?;
455 Ok(out)
456 }
457}
458#[cfg(test)]
459impl DescribedGroup {
460 #[must_use]
461 pub fn populated(version: i16) -> Self {
462 let mut m = Self::default();
463 if version >= 0 {
464 m.error_code = 1i16;
465 }
466 if version >= 0 {
467 m.error_message = Some("x".to_string());
468 }
469 if version >= 0 {
470 m.group_id = "x".to_string();
471 }
472 if version >= 0 {
473 m.group_state = "x".to_string();
474 }
475 if version >= 0 {
476 m.group_epoch = 1i32;
477 }
478 if version >= 0 {
479 m.assignment_epoch = 1i32;
480 }
481 if version >= 0 {
482 m.assignor_name = "x".to_string();
483 }
484 if version >= 0 {
485 m.members = vec![Member::populated(version)];
486 }
487 if version >= 0 {
488 m.authorized_operations = 1i32;
489 }
490 m
491 }
492}
493#[derive(Debug, Clone, PartialEq, Eq)]
494pub struct Member {
495 pub member_id: String,
496 pub instance_id: Option<String>,
497 pub rack_id: Option<String>,
498 pub member_epoch: i32,
499 pub client_id: String,
500 pub client_host: String,
501 pub subscribed_topic_names: Vec<String>,
502 pub subscribed_topic_regex: Option<String>,
503 pub assignment: super::common::consumer_group_describe_response::assignment::Assignment,
504 pub target_assignment: super::common::consumer_group_describe_response::assignment::Assignment,
505 pub member_type: i8,
506 pub unknown_tagged_fields: UnknownTaggedFields,
507}
508impl Default for Member {
509 fn default() -> Self {
510 Self {
511 member_id: String::new(),
512 instance_id: None,
513 rack_id: None,
514 member_epoch: 0i32,
515 client_id: String::new(),
516 client_host: String::new(),
517 subscribed_topic_names: Vec::new(),
518 subscribed_topic_regex: None,
519 assignment:
520 super::common::consumer_group_describe_response::assignment::Assignment::default(),
521 target_assignment:
522 super::common::consumer_group_describe_response::assignment::Assignment::default(),
523 member_type: -1i8,
524 unknown_tagged_fields: UnknownTaggedFields::default(),
525 }
526 }
527}
528impl Member {
529 fn encode_field_0<B: BufMut>(&self, buf: &mut B, version: i16, flex: bool) {
530 if version >= 0 {
531 if flex {
532 let () = put_compact_string(buf, &self.member_id);
533 } else {
534 let () = put_string(buf, &self.member_id);
535 }
536 }
537 }
538 fn encode_field_1<B: BufMut>(&self, buf: &mut B, version: i16, flex: bool) {
539 if version >= 0 {
540 if flex {
541 let () = put_compact_nullable_string(buf, self.instance_id.as_deref());
542 } else {
543 let () = put_nullable_string(buf, self.instance_id.as_deref());
544 }
545 }
546 }
547 fn encode_field_2<B: BufMut>(&self, buf: &mut B, version: i16, flex: bool) {
548 if version >= 0 {
549 if flex {
550 let () = put_compact_nullable_string(buf, self.rack_id.as_deref());
551 } else {
552 let () = put_nullable_string(buf, self.rack_id.as_deref());
553 }
554 }
555 }
556 fn encode_field_3<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
557 if version >= 0 {
558 put_i32(buf, self.member_epoch);
559 }
560 }
561 fn encode_field_4<B: BufMut>(&self, buf: &mut B, version: i16, flex: bool) {
562 if version >= 0 {
563 if flex {
564 let () = put_compact_string(buf, &self.client_id);
565 } else {
566 let () = put_string(buf, &self.client_id);
567 }
568 }
569 }
570 fn encode_field_5<B: BufMut>(&self, buf: &mut B, version: i16, flex: bool) {
571 if version >= 0 {
572 if flex {
573 let () = put_compact_string(buf, &self.client_host);
574 } else {
575 let () = put_string(buf, &self.client_host);
576 }
577 }
578 }
579 fn encode_field_6<B: BufMut>(&self, buf: &mut B, version: i16, flex: bool) {
580 if version >= 0 {
581 {
582 crate::primitives::array::put_array_len(
583 buf,
584 (self.subscribed_topic_names).len(),
585 flex,
586 );
587 for it in &self.subscribed_topic_names {
588 if flex {
589 let () = put_compact_string(buf, it);
590 } else {
591 let () = put_string(buf, it);
592 }
593 }
594 };
595 }
596 }
597 fn encode_field_7<B: BufMut>(&self, buf: &mut B, version: i16, flex: bool) {
598 if version >= 0 {
599 if flex {
600 let () = put_compact_nullable_string(buf, self.subscribed_topic_regex.as_deref());
601 } else {
602 let () = put_nullable_string(buf, self.subscribed_topic_regex.as_deref());
603 }
604 }
605 }
606 fn encode_field_8<B: BufMut>(
607 &self,
608 buf: &mut B,
609 version: i16,
610 _flex: bool,
611 ) -> Result<(), ProtocolError> {
612 if version >= 0 {
613 self.assignment.encode(buf, version)?;
614 }
615 Ok(())
616 }
617 fn encode_field_9<B: BufMut>(
618 &self,
619 buf: &mut B,
620 version: i16,
621 _flex: bool,
622 ) -> Result<(), ProtocolError> {
623 if version >= 0 {
624 self.target_assignment.encode(buf, version)?;
625 }
626 Ok(())
627 }
628 fn encode_field_10<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
629 if version >= 1 {
630 put_i8(buf, self.member_type);
631 }
632 }
633 fn encode_tagged_fields<B: BufMut>(&self, buf: &mut B, _version: i16, flex: bool) {
634 if flex {
635 let tagged = WriteTaggedFields::new();
636 tagged.write(buf, &self.unknown_tagged_fields);
637 }
638 }
639 fn decode_field_0<B: Buf>(
640 out: &mut Self,
641 buf: &mut B,
642 version: i16,
643 flex: bool,
644 ) -> Result<(), ProtocolError> {
645 if version >= 0 {
646 out.member_id = if flex {
647 get_compact_string_owned(buf)?
648 } else {
649 get_string_owned(buf)?
650 };
651 }
652 Ok(())
653 }
654 fn decode_field_1<B: Buf>(
655 out: &mut Self,
656 buf: &mut B,
657 version: i16,
658 flex: bool,
659 ) -> Result<(), ProtocolError> {
660 if version >= 0 {
661 out.instance_id = if flex {
662 get_compact_nullable_string_owned(buf)?
663 } else {
664 get_nullable_string_owned(buf)?
665 };
666 }
667 Ok(())
668 }
669 fn decode_field_2<B: Buf>(
670 out: &mut Self,
671 buf: &mut B,
672 version: i16,
673 flex: bool,
674 ) -> Result<(), ProtocolError> {
675 if version >= 0 {
676 out.rack_id = if flex {
677 get_compact_nullable_string_owned(buf)?
678 } else {
679 get_nullable_string_owned(buf)?
680 };
681 }
682 Ok(())
683 }
684 fn decode_field_3<B: Buf>(
685 out: &mut Self,
686 buf: &mut B,
687 version: i16,
688 _flex: bool,
689 ) -> Result<(), ProtocolError> {
690 if version >= 0 {
691 out.member_epoch = get_i32(buf)?;
692 }
693 Ok(())
694 }
695 fn decode_field_4<B: Buf>(
696 out: &mut Self,
697 buf: &mut B,
698 version: i16,
699 flex: bool,
700 ) -> Result<(), ProtocolError> {
701 if version >= 0 {
702 out.client_id = if flex {
703 get_compact_string_owned(buf)?
704 } else {
705 get_string_owned(buf)?
706 };
707 }
708 Ok(())
709 }
710 fn decode_field_5<B: Buf>(
711 out: &mut Self,
712 buf: &mut B,
713 version: i16,
714 flex: bool,
715 ) -> Result<(), ProtocolError> {
716 if version >= 0 {
717 out.client_host = if flex {
718 get_compact_string_owned(buf)?
719 } else {
720 get_string_owned(buf)?
721 };
722 }
723 Ok(())
724 }
725 fn decode_field_6<B: Buf>(
726 out: &mut Self,
727 buf: &mut B,
728 version: i16,
729 flex: bool,
730 ) -> Result<(), ProtocolError> {
731 if version >= 0 {
732 out.subscribed_topic_names = {
733 let n = crate::primitives::array::get_array_len(buf, flex)?;
734 let mut v = Vec::with_capacity(n);
735 for _ in 0..n {
736 v.push(if flex {
737 get_compact_string_owned(buf)?
738 } else {
739 get_string_owned(buf)?
740 });
741 }
742 v
743 };
744 }
745 Ok(())
746 }
747 fn decode_field_7<B: Buf>(
748 out: &mut Self,
749 buf: &mut B,
750 version: i16,
751 flex: bool,
752 ) -> Result<(), ProtocolError> {
753 if version >= 0 {
754 out.subscribed_topic_regex = if flex {
755 get_compact_nullable_string_owned(buf)?
756 } else {
757 get_nullable_string_owned(buf)?
758 };
759 }
760 Ok(())
761 }
762 fn decode_field_8<B: Buf>(
763 out: &mut Self,
764 buf: &mut B,
765 version: i16,
766 _flex: bool,
767 ) -> Result<(), ProtocolError> {
768 if version >= 0 {
769 out.assignment =
770 super::common::consumer_group_describe_response::assignment::Assignment::decode(
771 buf, version,
772 )?;
773 }
774 Ok(())
775 }
776 fn decode_field_9<B: Buf>(
777 out: &mut Self,
778 buf: &mut B,
779 version: i16,
780 _flex: bool,
781 ) -> Result<(), ProtocolError> {
782 if version >= 0 {
783 out.target_assignment =
784 super::common::consumer_group_describe_response::assignment::Assignment::decode(
785 buf, version,
786 )?;
787 }
788 Ok(())
789 }
790 fn decode_field_10<B: Buf>(
791 out: &mut Self,
792 buf: &mut B,
793 version: i16,
794 _flex: bool,
795 ) -> Result<(), ProtocolError> {
796 if version >= 1 {
797 out.member_type = get_i8(buf)?;
798 }
799 Ok(())
800 }
801 fn decode_tagged_fields<B: Buf>(
802 out: &mut Self,
803 buf: &mut B,
804 _version: i16,
805 flex: bool,
806 ) -> Result<(), ProtocolError> {
807 if flex {
808 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
809 }
810 Ok(())
811 }
812}
813impl Encode for Member {
814 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
815 let flex = version >= 0;
816 self.encode_field_0(buf, version, flex);
817 self.encode_field_1(buf, version, flex);
818 self.encode_field_2(buf, version, flex);
819 self.encode_field_3(buf, version, flex);
820 self.encode_field_4(buf, version, flex);
821 self.encode_field_5(buf, version, flex);
822 self.encode_field_6(buf, version, flex);
823 self.encode_field_7(buf, version, flex);
824 self.encode_field_8(buf, version, flex)?;
825 self.encode_field_9(buf, version, flex)?;
826 self.encode_field_10(buf, version, flex);
827 self.encode_tagged_fields(buf, version, flex);
828 Ok(())
829 }
830 fn encoded_len(&self, version: i16) -> usize {
831 let flex = version >= 0;
832 let mut n: usize = 0;
833 if version >= 0 {
834 n += if flex {
835 compact_string_len(&self.member_id)
836 } else {
837 string_len(&self.member_id)
838 };
839 }
840 if version >= 0 {
841 n += if flex {
842 compact_nullable_string_len(self.instance_id.as_deref())
843 } else {
844 nullable_string_len(self.instance_id.as_deref())
845 };
846 }
847 if version >= 0 {
848 n += if flex {
849 compact_nullable_string_len(self.rack_id.as_deref())
850 } else {
851 nullable_string_len(self.rack_id.as_deref())
852 };
853 }
854 if version >= 0 {
855 n += 4;
856 }
857 if version >= 0 {
858 n += if flex {
859 compact_string_len(&self.client_id)
860 } else {
861 string_len(&self.client_id)
862 };
863 }
864 if version >= 0 {
865 n += if flex {
866 compact_string_len(&self.client_host)
867 } else {
868 string_len(&self.client_host)
869 };
870 }
871 if version >= 0 {
872 n += {
873 let prefix = crate::primitives::array::array_len_prefix_len(
874 (self.subscribed_topic_names).len(),
875 flex,
876 );
877 let body: usize = (self.subscribed_topic_names)
878 .iter()
879 .map(|it| {
880 if flex {
881 compact_string_len(it)
882 } else {
883 string_len(it)
884 }
885 })
886 .sum();
887 prefix + body
888 };
889 }
890 if version >= 0 {
891 n += if flex {
892 compact_nullable_string_len(self.subscribed_topic_regex.as_deref())
893 } else {
894 nullable_string_len(self.subscribed_topic_regex.as_deref())
895 };
896 }
897 if version >= 0 {
898 n += self.assignment.encoded_len(version);
899 }
900 if version >= 0 {
901 n += self.target_assignment.encoded_len(version);
902 }
903 if version >= 1 {
904 n += 1;
905 }
906 if flex {
907 let known_pairs: Vec<(u32, usize)> = Vec::new();
908 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
909 }
910 n
911 }
912}
913impl Decode<'_> for Member {
914 fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
915 let flex = version >= 0;
916 let mut out = Self::default();
917 Self::decode_field_0(&mut out, buf, version, flex)?;
918 Self::decode_field_1(&mut out, buf, version, flex)?;
919 Self::decode_field_2(&mut out, buf, version, flex)?;
920 Self::decode_field_3(&mut out, buf, version, flex)?;
921 Self::decode_field_4(&mut out, buf, version, flex)?;
922 Self::decode_field_5(&mut out, buf, version, flex)?;
923 Self::decode_field_6(&mut out, buf, version, flex)?;
924 Self::decode_field_7(&mut out, buf, version, flex)?;
925 Self::decode_field_8(&mut out, buf, version, flex)?;
926 Self::decode_field_9(&mut out, buf, version, flex)?;
927 Self::decode_field_10(&mut out, buf, version, flex)?;
928 Self::decode_tagged_fields(&mut out, buf, version, flex)?;
929 Ok(out)
930 }
931}
932#[cfg(test)]
933impl Member {
934 #[must_use]
935 pub fn populated(version: i16) -> Self {
936 let mut m = Self::default();
937 if version >= 0 {
938 m.member_id = "x".to_string();
939 }
940 if version >= 0 {
941 m.instance_id = Some("x".to_string());
942 }
943 if version >= 0 {
944 m.rack_id = Some("x".to_string());
945 }
946 if version >= 0 {
947 m.member_epoch = 1i32;
948 }
949 if version >= 0 {
950 m.client_id = "x".to_string();
951 }
952 if version >= 0 {
953 m.client_host = "x".to_string();
954 }
955 if version >= 0 {
956 m.subscribed_topic_names = vec!["x".to_string()];
957 }
958 if version >= 0 {
959 m.subscribed_topic_regex = Some("x".to_string());
960 }
961 if version >= 0 {
962 m.assignment =
963 super::common::consumer_group_describe_response::assignment::Assignment::populated(
964 version,
965 );
966 }
967 if version >= 0 {
968 m.target_assignment =
969 super::common::consumer_group_describe_response::assignment::Assignment::populated(
970 version,
971 );
972 }
973 if version >= 1 {
974 m.member_type = 1i8;
975 }
976 m
977 }
978}
979#[must_use]
982#[allow(unused_comparisons)]
983pub fn default_json(_version: i16) -> ::serde_json::Value {
984 let mut obj = ::serde_json::Map::new();
985 obj.insert("throttleTimeMs".to_string(), ::serde_json::json!(0));
986 obj.insert("groups".to_string(), ::serde_json::Value::Array(vec![]));
987 ::serde_json::Value::Object(obj)
988}