1use bytes::BufMut;
3
4use crate::{
5 DecodeBorrow, Encode, ProtocolError, UnknownTaggedFields,
6 primitives::{
7 fixed::{get_bool, get_i8, get_i16, get_i32, put_bool, put_i8, put_i16, put_i32},
8 string_bytes::{
9 compact_nullable_string_len, compact_string_len, nullable_string_len,
10 put_compact_nullable_string, put_compact_string, put_nullable_string, put_string,
11 string_len,
12 },
13 string_bytes_borrowed::{
14 get_compact_nullable_string_borrowed, get_compact_string_borrowed,
15 get_nullable_string_borrowed, get_string_borrowed,
16 },
17 },
18 tagged_fields::{WriteTaggedFields, read_tagged_fields, tagged_fields_len},
19};
20pub const API_KEY: i16 = 89;
21pub const MIN_VERSION: i16 = 0;
22pub const MAX_VERSION: i16 = 1;
23pub const FLEXIBLE_MIN: i16 = 0;
24#[inline]
25#[must_use]
26pub fn is_flexible(version: i16) -> bool {
27 version >= FLEXIBLE_MIN
28}
29#[derive(Debug, Clone, PartialEq, Eq, Default)]
30pub struct StreamsGroupDescribeResponse<'a> {
31 pub throttle_time_ms: i32,
32 pub groups: Vec<DescribedGroup<'a>>,
33 pub unknown_tagged_fields: UnknownTaggedFields,
34}
35impl StreamsGroupDescribeResponse<'_> {
36 #[must_use]
40 pub fn to_owned(
41 &self,
42 ) -> crate::owned::streams_group_describe_response::StreamsGroupDescribeResponse {
43 crate::owned::streams_group_describe_response::StreamsGroupDescribeResponse {
44 throttle_time_ms: (self.throttle_time_ms),
45 groups: (self.groups).iter().map(DescribedGroup::to_owned).collect(),
46 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
47 }
48 }
49}
50impl Encode for StreamsGroupDescribeResponse<'_> {
51 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
52 if !(MIN_VERSION..=MAX_VERSION).contains(&version) {
53 return Err(ProtocolError::UnsupportedVersion {
54 api_key: API_KEY,
55 version,
56 });
57 }
58 let flex = is_flexible(version);
59 if version >= 0 {
60 put_i32(buf, self.throttle_time_ms);
61 }
62 if version >= 0 {
63 {
64 crate::primitives::array::put_array_len(buf, (self.groups).len(), flex);
65 for it in &self.groups {
66 it.encode(buf, version)?;
67 }
68 };
69 }
70 if flex {
71 let tagged = WriteTaggedFields::new();
72 tagged.write(buf, &self.unknown_tagged_fields);
73 }
74 Ok(())
75 }
76 fn encoded_len(&self, version: i16) -> usize {
77 let flex = is_flexible(version);
78 let mut n: usize = 0;
79 if version >= 0 {
80 n += 4;
81 }
82 if version >= 0 {
83 n += {
84 let prefix =
85 crate::primitives::array::array_len_prefix_len((self.groups).len(), flex);
86 let body: usize = (self.groups).iter().map(|it| it.encoded_len(version)).sum();
87 prefix + body
88 };
89 }
90 if flex {
91 let known_pairs: Vec<(u32, usize)> = Vec::new();
92 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
93 }
94 n
95 }
96}
97impl<'de> DecodeBorrow<'de> for StreamsGroupDescribeResponse<'de> {
98 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
99 if !(MIN_VERSION..=MAX_VERSION).contains(&version) {
100 return Err(ProtocolError::UnsupportedVersion {
101 api_key: API_KEY,
102 version,
103 });
104 }
105 let flex = is_flexible(version);
106 let mut out = Self::default();
107 if version >= 0 {
108 out.throttle_time_ms = get_i32(buf)?;
109 }
110 if version >= 0 {
111 out.groups = {
112 let n = crate::primitives::array::get_array_len(buf, flex)?;
113 let mut v = Vec::with_capacity(n);
114 for _ in 0..n {
115 v.push(DescribedGroup::decode_borrow(buf, version)?);
116 }
117 v
118 };
119 }
120 if flex {
121 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
122 }
123 Ok(out)
124 }
125}
126#[cfg(test)]
127impl StreamsGroupDescribeResponse<'_> {
128 #[must_use]
129 pub fn populated(version: i16) -> Self {
130 let mut m = Self::default();
131 if version >= 0 {
132 m.throttle_time_ms = 1i32;
133 }
134 if version >= 0 {
135 m.groups = vec![DescribedGroup::populated(version)];
136 }
137 m
138 }
139}
140#[derive(Debug, Clone, PartialEq, Eq)]
141pub struct DescribedGroup<'a> {
142 pub error_code: i16,
143 pub error_message: Option<&'a str>,
144 pub group_id: &'a str,
145 pub group_state: &'a str,
146 pub group_epoch: i32,
147 pub assignment_epoch: i32,
148 pub topology: Option<Topology<'a>>,
149 pub members: Vec<Member<'a>>,
150 pub authorized_operations: i32,
151 pub topology_description: Option<
152 super::common::streams_group_describe_response::topology_description::TopologyDescription<
153 'a,
154 >,
155 >,
156 pub topology_description_status: i8,
157 pub assignor_name: Option<&'a str>,
158 pub unknown_tagged_fields: UnknownTaggedFields,
159}
160impl Default for DescribedGroup<'_> {
161 fn default() -> Self {
162 Self {
163 error_code: 0i16,
164 error_message: None,
165 group_id: "",
166 group_state: "",
167 group_epoch: 0i32,
168 assignment_epoch: 0i32,
169 topology: None,
170 members: Vec::new(),
171 authorized_operations: -2_147_483_648i32,
172 topology_description: None,
173 topology_description_status: 0i8,
174 assignor_name: None,
175 unknown_tagged_fields: UnknownTaggedFields::default(),
176 }
177 }
178}
179impl<'a> DescribedGroup<'a> {
180 #[must_use]
184 pub fn to_owned(&self) -> crate::owned::streams_group_describe_response::DescribedGroup {
185 crate::owned::streams_group_describe_response::DescribedGroup {
186 error_code: (self.error_code),
187 error_message: (self.error_message).map(std::string::ToString::to_string),
188 group_id: (self.group_id).to_string(),
189 group_state: (self.group_state).to_string(),
190 group_epoch: (self.group_epoch),
191 assignment_epoch: (self.assignment_epoch),
192 topology: (self.topology).as_ref().map(Topology::to_owned),
193 members: (self.members).iter().map(Member::to_owned).collect(),
194 authorized_operations: (self.authorized_operations),
195 topology_description: (self.topology_description).as_ref().map(super::common::streams_group_describe_response::topology_description::TopologyDescription::to_owned),
196 topology_description_status: (self.topology_description_status),
197 assignor_name: (self.assignor_name).map(std::string::ToString::to_string),
198 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
199 }
200 }
201 fn encode_field_0<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
202 if version >= 0 {
203 put_i16(buf, self.error_code);
204 }
205 }
206 fn encode_field_1<B: BufMut>(&self, buf: &mut B, version: i16, flex: bool) {
207 if version >= 0 {
208 if flex {
209 let () = put_compact_nullable_string(buf, self.error_message);
210 } else {
211 let () = put_nullable_string(buf, self.error_message);
212 }
213 }
214 }
215 fn encode_field_2<B: BufMut>(&self, buf: &mut B, version: i16, flex: bool) {
216 if version >= 0 {
217 if flex {
218 let () = put_compact_string(buf, self.group_id);
219 } else {
220 let () = put_string(buf, self.group_id);
221 }
222 }
223 }
224 fn encode_field_3<B: BufMut>(&self, buf: &mut B, version: i16, flex: bool) {
225 if version >= 0 {
226 if flex {
227 let () = put_compact_string(buf, self.group_state);
228 } else {
229 let () = put_string(buf, self.group_state);
230 }
231 }
232 }
233 fn encode_field_4<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
234 if version >= 0 {
235 put_i32(buf, self.group_epoch);
236 }
237 }
238 fn encode_field_5<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
239 if version >= 0 {
240 put_i32(buf, self.assignment_epoch);
241 }
242 }
243 fn encode_field_6<B: BufMut>(
244 &self,
245 buf: &mut B,
246 version: i16,
247 _flex: bool,
248 ) -> Result<(), ProtocolError> {
249 if version >= 0 {
250 match &self.topology {
251 None => {
252 buf.put_i8(-1);
253 }
254 Some(v) => {
255 buf.put_i8(1);
256 v.encode(buf, version)?;
257 }
258 }
259 }
260 Ok(())
261 }
262 fn encode_field_7<B: BufMut>(
263 &self,
264 buf: &mut B,
265 version: i16,
266 flex: bool,
267 ) -> Result<(), ProtocolError> {
268 if version >= 0 {
269 {
270 crate::primitives::array::put_array_len(buf, (self.members).len(), flex);
271 for it in &self.members {
272 it.encode(buf, version)?;
273 }
274 };
275 }
276 Ok(())
277 }
278 fn encode_field_8<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
279 if version >= 0 {
280 put_i32(buf, self.authorized_operations);
281 }
282 }
283 fn encode_field_9<B: BufMut>(
284 &self,
285 buf: &mut B,
286 version: i16,
287 _flex: bool,
288 ) -> Result<(), ProtocolError> {
289 if version >= 1 {
290 match &self.topology_description {
291 None => {
292 buf.put_i8(-1);
293 }
294 Some(v) => {
295 buf.put_i8(1);
296 v.encode(buf, version)?;
297 }
298 }
299 }
300 Ok(())
301 }
302 fn encode_field_10<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
303 if version >= 1 {
304 put_i8(buf, self.topology_description_status);
305 }
306 }
307 fn encode_field_11<B: BufMut>(&self, buf: &mut B, version: i16, flex: bool) {
308 if version >= 1 {
309 if flex {
310 let () = put_compact_nullable_string(buf, self.assignor_name);
311 } else {
312 let () = put_nullable_string(buf, self.assignor_name);
313 }
314 }
315 }
316 fn encode_tagged_fields<B: BufMut>(&self, buf: &mut B, _version: i16, flex: bool) {
317 if flex {
318 let tagged = WriteTaggedFields::new();
319 tagged.write(buf, &self.unknown_tagged_fields);
320 }
321 }
322 fn decode_field_0(
323 out: &mut Self,
324 buf: &mut &'a [u8],
325 version: i16,
326 _flex: bool,
327 ) -> Result<(), ProtocolError> {
328 if version >= 0 {
329 out.error_code = get_i16(buf)?;
330 }
331 Ok(())
332 }
333 fn decode_field_1(
334 out: &mut Self,
335 buf: &mut &'a [u8],
336 version: i16,
337 flex: bool,
338 ) -> Result<(), ProtocolError> {
339 if version >= 0 {
340 out.error_message = if flex {
341 get_compact_nullable_string_borrowed(buf)?
342 } else {
343 get_nullable_string_borrowed(buf)?
344 };
345 }
346 Ok(())
347 }
348 fn decode_field_2(
349 out: &mut Self,
350 buf: &mut &'a [u8],
351 version: i16,
352 flex: bool,
353 ) -> Result<(), ProtocolError> {
354 if version >= 0 {
355 out.group_id = if flex {
356 get_compact_string_borrowed(buf)?
357 } else {
358 get_string_borrowed(buf)?
359 };
360 }
361 Ok(())
362 }
363 fn decode_field_3(
364 out: &mut Self,
365 buf: &mut &'a [u8],
366 version: i16,
367 flex: bool,
368 ) -> Result<(), ProtocolError> {
369 if version >= 0 {
370 out.group_state = if flex {
371 get_compact_string_borrowed(buf)?
372 } else {
373 get_string_borrowed(buf)?
374 };
375 }
376 Ok(())
377 }
378 fn decode_field_4(
379 out: &mut Self,
380 buf: &mut &'a [u8],
381 version: i16,
382 _flex: bool,
383 ) -> Result<(), ProtocolError> {
384 if version >= 0 {
385 out.group_epoch = get_i32(buf)?;
386 }
387 Ok(())
388 }
389 fn decode_field_5(
390 out: &mut Self,
391 buf: &mut &'a [u8],
392 version: i16,
393 _flex: bool,
394 ) -> Result<(), ProtocolError> {
395 if version >= 0 {
396 out.assignment_epoch = get_i32(buf)?;
397 }
398 Ok(())
399 }
400 fn decode_field_6(
401 out: &mut Self,
402 buf: &mut &'a [u8],
403 version: i16,
404 _flex: bool,
405 ) -> Result<(), ProtocolError> {
406 if version >= 0 {
407 out.topology = if get_i8(buf)? < 0 {
408 None
409 } else {
410 Some(Topology::decode_borrow(buf, version)?)
411 };
412 }
413 Ok(())
414 }
415 fn decode_field_7(
416 out: &mut Self,
417 buf: &mut &'a [u8],
418 version: i16,
419 flex: bool,
420 ) -> Result<(), ProtocolError> {
421 if version >= 0 {
422 out.members = {
423 let n = crate::primitives::array::get_array_len(buf, flex)?;
424 let mut v = Vec::with_capacity(n);
425 for _ in 0..n {
426 v.push(Member::decode_borrow(buf, version)?);
427 }
428 v
429 };
430 }
431 Ok(())
432 }
433 fn decode_field_8(
434 out: &mut Self,
435 buf: &mut &'a [u8],
436 version: i16,
437 _flex: bool,
438 ) -> Result<(), ProtocolError> {
439 if version >= 0 {
440 out.authorized_operations = get_i32(buf)?;
441 }
442 Ok(())
443 }
444 fn decode_field_9(
445 out: &mut Self,
446 buf: &mut &'a [u8],
447 version: i16,
448 _flex: bool,
449 ) -> Result<(), ProtocolError> {
450 if version >= 1 {
451 out.topology_description = if get_i8(buf)? < 0 {
452 None
453 } else {
454 Some(super::common::streams_group_describe_response::topology_description::TopologyDescription::decode_borrow(buf, version)?)
455 };
456 }
457 Ok(())
458 }
459 fn decode_field_10(
460 out: &mut Self,
461 buf: &mut &'a [u8],
462 version: i16,
463 _flex: bool,
464 ) -> Result<(), ProtocolError> {
465 if version >= 1 {
466 out.topology_description_status = get_i8(buf)?;
467 }
468 Ok(())
469 }
470 fn decode_field_11(
471 out: &mut Self,
472 buf: &mut &'a [u8],
473 version: i16,
474 flex: bool,
475 ) -> Result<(), ProtocolError> {
476 if version >= 1 {
477 out.assignor_name = if flex {
478 get_compact_nullable_string_borrowed(buf)?
479 } else {
480 get_nullable_string_borrowed(buf)?
481 };
482 }
483 Ok(())
484 }
485 fn decode_tagged_fields(
486 out: &mut Self,
487 buf: &mut &'a [u8],
488 _version: i16,
489 flex: bool,
490 ) -> Result<(), ProtocolError> {
491 if flex {
492 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
493 }
494 Ok(())
495 }
496}
497impl Encode for DescribedGroup<'_> {
498 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
499 let flex = version >= 0;
500 self.encode_field_0(buf, version, flex);
501 self.encode_field_1(buf, version, flex);
502 self.encode_field_2(buf, version, flex);
503 self.encode_field_3(buf, version, flex);
504 self.encode_field_4(buf, version, flex);
505 self.encode_field_5(buf, version, flex);
506 self.encode_field_6(buf, version, flex)?;
507 self.encode_field_7(buf, version, flex)?;
508 self.encode_field_8(buf, version, flex);
509 self.encode_field_9(buf, version, flex)?;
510 self.encode_field_10(buf, version, flex);
511 self.encode_field_11(buf, version, flex);
512 self.encode_tagged_fields(buf, version, flex);
513 Ok(())
514 }
515 fn encoded_len(&self, version: i16) -> usize {
516 let flex = version >= 0;
517 let mut n: usize = 0;
518 if version >= 0 {
519 n += 2;
520 }
521 if version >= 0 {
522 n += if flex {
523 compact_nullable_string_len(self.error_message)
524 } else {
525 nullable_string_len(self.error_message)
526 };
527 }
528 if version >= 0 {
529 n += if flex {
530 compact_string_len(self.group_id)
531 } else {
532 string_len(self.group_id)
533 };
534 }
535 if version >= 0 {
536 n += if flex {
537 compact_string_len(self.group_state)
538 } else {
539 string_len(self.group_state)
540 };
541 }
542 if version >= 0 {
543 n += 4;
544 }
545 if version >= 0 {
546 n += 4;
547 }
548 if version >= 0 {
549 n += 1 + self.topology.as_ref().map_or(0, |v| v.encoded_len(version));
550 }
551 if version >= 0 {
552 n += {
553 let prefix =
554 crate::primitives::array::array_len_prefix_len((self.members).len(), flex);
555 let body: usize = (self.members)
556 .iter()
557 .map(|it| it.encoded_len(version))
558 .sum();
559 prefix + body
560 };
561 }
562 if version >= 0 {
563 n += 4;
564 }
565 if version >= 1 {
566 n += 1 + self
567 .topology_description
568 .as_ref()
569 .map_or(0, |v| v.encoded_len(version));
570 }
571 if version >= 1 {
572 n += 1;
573 }
574 if version >= 1 {
575 n += if flex {
576 compact_nullable_string_len(self.assignor_name)
577 } else {
578 nullable_string_len(self.assignor_name)
579 };
580 }
581 if flex {
582 let known_pairs: Vec<(u32, usize)> = Vec::new();
583 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
584 }
585 n
586 }
587}
588impl<'de> DecodeBorrow<'de> for DescribedGroup<'de> {
589 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
590 let flex = version >= 0;
591 let mut out = Self::default();
592 Self::decode_field_0(&mut out, buf, version, flex)?;
593 Self::decode_field_1(&mut out, buf, version, flex)?;
594 Self::decode_field_2(&mut out, buf, version, flex)?;
595 Self::decode_field_3(&mut out, buf, version, flex)?;
596 Self::decode_field_4(&mut out, buf, version, flex)?;
597 Self::decode_field_5(&mut out, buf, version, flex)?;
598 Self::decode_field_6(&mut out, buf, version, flex)?;
599 Self::decode_field_7(&mut out, buf, version, flex)?;
600 Self::decode_field_8(&mut out, buf, version, flex)?;
601 Self::decode_field_9(&mut out, buf, version, flex)?;
602 Self::decode_field_10(&mut out, buf, version, flex)?;
603 Self::decode_field_11(&mut out, buf, version, flex)?;
604 Self::decode_tagged_fields(&mut out, buf, version, flex)?;
605 Ok(out)
606 }
607}
608#[cfg(test)]
609impl DescribedGroup<'_> {
610 #[must_use]
611 pub fn populated(version: i16) -> Self {
612 let mut m = Self::default();
613 if version >= 0 {
614 m.error_code = 1i16;
615 }
616 if version >= 0 {
617 m.error_message = Some("x");
618 }
619 if version >= 0 {
620 m.group_id = "x";
621 }
622 if version >= 0 {
623 m.group_state = "x";
624 }
625 if version >= 0 {
626 m.group_epoch = 1i32;
627 }
628 if version >= 0 {
629 m.assignment_epoch = 1i32;
630 }
631 if version >= 0 {
632 m.topology = Some(Topology::populated(version));
633 }
634 if version >= 0 {
635 m.members = vec![Member::populated(version)];
636 }
637 if version >= 0 {
638 m.authorized_operations = 1i32;
639 }
640 if version >= 1 {
641 m.topology_description = Some(super::common::streams_group_describe_response::topology_description::TopologyDescription::populated(version));
642 }
643 if version >= 1 {
644 m.topology_description_status = 1i8;
645 }
646 if version >= 1 {
647 m.assignor_name = Some("x");
648 }
649 m
650 }
651}
652#[derive(Debug, Clone, PartialEq, Eq, Default)]
653pub struct Topology<'a> {
654 pub epoch: i32,
655 pub subtopologies: Option<Vec<Subtopology<'a>>>,
656 pub unknown_tagged_fields: UnknownTaggedFields,
657}
658impl Topology<'_> {
659 #[must_use]
663 pub fn to_owned(&self) -> crate::owned::streams_group_describe_response::Topology {
664 crate::owned::streams_group_describe_response::Topology {
665 epoch: (self.epoch),
666 subtopologies: (self.subtopologies)
667 .as_ref()
668 .map(|v| v.iter().map(Subtopology::to_owned).collect()),
669 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
670 }
671 }
672}
673impl Encode for Topology<'_> {
674 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
675 let flex = version >= 0;
676 if version >= 0 {
677 put_i32(buf, self.epoch);
678 }
679 if version >= 0 {
680 {
681 let len = (self.subtopologies).as_ref().map(Vec::len);
682 crate::primitives::array::put_nullable_array_len(buf, len, flex);
683 if let Some(v) = &self.subtopologies {
684 for it in v {
685 it.encode(buf, version)?;
686 }
687 }
688 };
689 }
690 if flex {
691 let tagged = WriteTaggedFields::new();
692 tagged.write(buf, &self.unknown_tagged_fields);
693 }
694 Ok(())
695 }
696 fn encoded_len(&self, version: i16) -> usize {
697 let flex = version >= 0;
698 let mut n: usize = 0;
699 if version >= 0 {
700 n += 4;
701 }
702 if version >= 0 {
703 n += {
704 let opt: Option<&Vec<_>> = (self.subtopologies).as_ref();
705 let prefix = crate::primitives::array::nullable_array_len_prefix_len(
706 opt.map(std::vec::Vec::len),
707 flex,
708 );
709 let body: usize =
710 opt.map_or(0, |v| v.iter().map(|it| it.encoded_len(version)).sum());
711 prefix + body
712 };
713 }
714 if flex {
715 let known_pairs: Vec<(u32, usize)> = Vec::new();
716 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
717 }
718 n
719 }
720}
721impl<'de> DecodeBorrow<'de> for Topology<'de> {
722 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
723 let flex = version >= 0;
724 let mut out = Self::default();
725 if version >= 0 {
726 out.epoch = get_i32(buf)?;
727 }
728 if version >= 0 {
729 out.subtopologies = {
730 let opt = crate::primitives::array::get_nullable_array_len(buf, flex)?;
731 match opt {
732 None => None,
733 Some(n) => {
734 let mut v = Vec::with_capacity(n);
735 for _ in 0..n {
736 v.push(Subtopology::decode_borrow(buf, version)?);
737 }
738 Some(v)
739 }
740 }
741 };
742 }
743 if flex {
744 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
745 }
746 Ok(out)
747 }
748}
749#[cfg(test)]
750impl Topology<'_> {
751 #[must_use]
752 pub fn populated(version: i16) -> Self {
753 let mut m = Self::default();
754 if version >= 0 {
755 m.epoch = 1i32;
756 }
757 if version >= 0 {
758 m.subtopologies = Some(vec![Subtopology::populated(version)]);
759 }
760 m
761 }
762}
763#[derive(Debug, Clone, PartialEq, Eq, Default)]
764pub struct Subtopology<'a> {
765 pub subtopology_id: &'a str,
766 pub source_topics: Vec<&'a str>,
767 pub repartition_sink_topics: Vec<&'a str>,
768 pub state_changelog_topics:
769 Vec<super::common::streams_group_describe_response::topic_info::TopicInfo<'a>>,
770 pub repartition_source_topics:
771 Vec<super::common::streams_group_describe_response::topic_info::TopicInfo<'a>>,
772 pub unknown_tagged_fields: UnknownTaggedFields,
773}
774impl Subtopology<'_> {
775 #[must_use]
779 pub fn to_owned(&self) -> crate::owned::streams_group_describe_response::Subtopology {
780 crate::owned::streams_group_describe_response::Subtopology {
781 subtopology_id: (self.subtopology_id).to_string(),
782 source_topics: (self.source_topics)
783 .iter()
784 .map(std::string::ToString::to_string)
785 .collect(),
786 repartition_sink_topics: (self.repartition_sink_topics)
787 .iter()
788 .map(std::string::ToString::to_string)
789 .collect(),
790 state_changelog_topics: (self.state_changelog_topics)
791 .iter()
792 .map(
793 super::common::streams_group_describe_response::topic_info::TopicInfo::to_owned,
794 )
795 .collect(),
796 repartition_source_topics: (self.repartition_source_topics)
797 .iter()
798 .map(
799 super::common::streams_group_describe_response::topic_info::TopicInfo::to_owned,
800 )
801 .collect(),
802 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
803 }
804 }
805}
806impl Encode for Subtopology<'_> {
807 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
808 let flex = version >= 0;
809 if version >= 0 {
810 if flex {
811 let () = put_compact_string(buf, self.subtopology_id);
812 } else {
813 let () = put_string(buf, self.subtopology_id);
814 }
815 }
816 if version >= 0 {
817 {
818 crate::primitives::array::put_array_len(buf, (self.source_topics).len(), flex);
819 for it in &self.source_topics {
820 if flex {
821 let () = put_compact_string(buf, it);
822 } else {
823 let () = put_string(buf, it);
824 }
825 }
826 };
827 }
828 if version >= 0 {
829 {
830 crate::primitives::array::put_array_len(
831 buf,
832 (self.repartition_sink_topics).len(),
833 flex,
834 );
835 for it in &self.repartition_sink_topics {
836 if flex {
837 let () = put_compact_string(buf, it);
838 } else {
839 let () = put_string(buf, it);
840 }
841 }
842 };
843 }
844 if version >= 0 {
845 {
846 crate::primitives::array::put_array_len(
847 buf,
848 (self.state_changelog_topics).len(),
849 flex,
850 );
851 for it in &self.state_changelog_topics {
852 it.encode(buf, version)?;
853 }
854 };
855 }
856 if version >= 0 {
857 {
858 crate::primitives::array::put_array_len(
859 buf,
860 (self.repartition_source_topics).len(),
861 flex,
862 );
863 for it in &self.repartition_source_topics {
864 it.encode(buf, version)?;
865 }
866 };
867 }
868 if flex {
869 let tagged = WriteTaggedFields::new();
870 tagged.write(buf, &self.unknown_tagged_fields);
871 }
872 Ok(())
873 }
874 fn encoded_len(&self, version: i16) -> usize {
875 let flex = version >= 0;
876 let mut n: usize = 0;
877 if version >= 0 {
878 n += if flex {
879 compact_string_len(self.subtopology_id)
880 } else {
881 string_len(self.subtopology_id)
882 };
883 }
884 if version >= 0 {
885 n += {
886 let prefix = crate::primitives::array::array_len_prefix_len(
887 (self.source_topics).len(),
888 flex,
889 );
890 let body: usize = (self.source_topics)
891 .iter()
892 .map(|it| {
893 if flex {
894 compact_string_len(it)
895 } else {
896 string_len(it)
897 }
898 })
899 .sum();
900 prefix + body
901 };
902 }
903 if version >= 0 {
904 n += {
905 let prefix = crate::primitives::array::array_len_prefix_len(
906 (self.repartition_sink_topics).len(),
907 flex,
908 );
909 let body: usize = (self.repartition_sink_topics)
910 .iter()
911 .map(|it| {
912 if flex {
913 compact_string_len(it)
914 } else {
915 string_len(it)
916 }
917 })
918 .sum();
919 prefix + body
920 };
921 }
922 if version >= 0 {
923 n += {
924 let prefix = crate::primitives::array::array_len_prefix_len(
925 (self.state_changelog_topics).len(),
926 flex,
927 );
928 let body: usize = (self.state_changelog_topics)
929 .iter()
930 .map(|it| it.encoded_len(version))
931 .sum();
932 prefix + body
933 };
934 }
935 if version >= 0 {
936 n += {
937 let prefix = crate::primitives::array::array_len_prefix_len(
938 (self.repartition_source_topics).len(),
939 flex,
940 );
941 let body: usize = (self.repartition_source_topics)
942 .iter()
943 .map(|it| it.encoded_len(version))
944 .sum();
945 prefix + body
946 };
947 }
948 if flex {
949 let known_pairs: Vec<(u32, usize)> = Vec::new();
950 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
951 }
952 n
953 }
954}
955impl<'de> DecodeBorrow<'de> for Subtopology<'de> {
956 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
957 let flex = version >= 0;
958 let mut out = Self::default();
959 if version >= 0 {
960 out.subtopology_id = if flex {
961 get_compact_string_borrowed(buf)?
962 } else {
963 get_string_borrowed(buf)?
964 };
965 }
966 if version >= 0 {
967 out.source_topics = {
968 let n = crate::primitives::array::get_array_len(buf, flex)?;
969 let mut v = Vec::with_capacity(n);
970 for _ in 0..n {
971 v.push(if flex {
972 get_compact_string_borrowed(buf)?
973 } else {
974 get_string_borrowed(buf)?
975 });
976 }
977 v
978 };
979 }
980 if version >= 0 {
981 out.repartition_sink_topics = {
982 let n = crate::primitives::array::get_array_len(buf, flex)?;
983 let mut v = Vec::with_capacity(n);
984 for _ in 0..n {
985 v.push(if flex {
986 get_compact_string_borrowed(buf)?
987 } else {
988 get_string_borrowed(buf)?
989 });
990 }
991 v
992 };
993 }
994 if version >= 0 {
995 out.state_changelog_topics = {
996 let n = crate::primitives::array::get_array_len(buf, flex)?;
997 let mut v = Vec::with_capacity(n);
998 for _ in 0..n {
999 v.push(super::common::streams_group_describe_response::topic_info::TopicInfo::decode_borrow(buf, version)?);
1000 }
1001 v
1002 };
1003 }
1004 if version >= 0 {
1005 out.repartition_source_topics = {
1006 let n = crate::primitives::array::get_array_len(buf, flex)?;
1007 let mut v = Vec::with_capacity(n);
1008 for _ in 0..n {
1009 v.push(super::common::streams_group_describe_response::topic_info::TopicInfo::decode_borrow(buf, version)?);
1010 }
1011 v
1012 };
1013 }
1014 if flex {
1015 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
1016 }
1017 Ok(out)
1018 }
1019}
1020#[cfg(test)]
1021impl Subtopology<'_> {
1022 #[must_use]
1023 pub fn populated(version: i16) -> Self {
1024 let mut m = Self::default();
1025 if version >= 0 {
1026 m.subtopology_id = "x";
1027 }
1028 if version >= 0 {
1029 m.source_topics = vec!["x"];
1030 }
1031 if version >= 0 {
1032 m.repartition_sink_topics = vec!["x"];
1033 }
1034 if version >= 0 {
1035 m.state_changelog_topics = vec![
1036 super::common::streams_group_describe_response::topic_info::TopicInfo::populated(
1037 version,
1038 ),
1039 ];
1040 }
1041 if version >= 0 {
1042 m.repartition_source_topics = vec![
1043 super::common::streams_group_describe_response::topic_info::TopicInfo::populated(
1044 version,
1045 ),
1046 ];
1047 }
1048 m
1049 }
1050}
1051#[derive(Debug, Clone, PartialEq, Eq, Default)]
1052pub struct Member<'a> {
1053 pub member_id: &'a str,
1054 pub member_epoch: i32,
1055 pub instance_id: Option<&'a str>,
1056 pub rack_id: Option<&'a str>,
1057 pub client_id: &'a str,
1058 pub client_host: &'a str,
1059 pub topology_epoch: i32,
1060 pub process_id: &'a str,
1061 pub user_endpoint:
1062 Option<super::common::streams_group_describe_response::endpoint::Endpoint<'a>>,
1063 pub client_tags: Vec<super::common::streams_group_describe_response::key_value::KeyValue<'a>>,
1064 pub task_offsets:
1065 Vec<super::common::streams_group_describe_response::task_offset::TaskOffset<'a>>,
1066 pub task_end_offsets:
1067 Vec<super::common::streams_group_describe_response::task_offset::TaskOffset<'a>>,
1068 pub assignment: super::common::streams_group_describe_response::assignment::Assignment<'a>,
1069 pub target_assignment:
1070 super::common::streams_group_describe_response::assignment::Assignment<'a>,
1071 pub is_classic: bool,
1072 pub unknown_tagged_fields: UnknownTaggedFields,
1073}
1074impl<'a> Member<'a> {
1075 #[must_use]
1079 pub fn to_owned(&self) -> crate::owned::streams_group_describe_response::Member {
1080 crate::owned::streams_group_describe_response::Member {
1081 member_id: (self.member_id).to_string(),
1082 member_epoch: (self.member_epoch),
1083 instance_id: (self.instance_id).map(std::string::ToString::to_string),
1084 rack_id: (self.rack_id).map(std::string::ToString::to_string),
1085 client_id: (self.client_id).to_string(),
1086 client_host: (self.client_host).to_string(),
1087 topology_epoch: (self.topology_epoch),
1088 process_id: (self.process_id).to_string(),
1089 user_endpoint: (self.user_endpoint).as_ref().map(super::common::streams_group_describe_response::endpoint::Endpoint::to_owned),
1090 client_tags: (self.client_tags).iter().map(super::common::streams_group_describe_response::key_value::KeyValue::to_owned).collect(),
1091 task_offsets: (self.task_offsets).iter().map(super::common::streams_group_describe_response::task_offset::TaskOffset::to_owned).collect(),
1092 task_end_offsets: (self.task_end_offsets).iter().map(super::common::streams_group_describe_response::task_offset::TaskOffset::to_owned).collect(),
1093 assignment: (self.assignment).to_owned(),
1094 target_assignment: (self.target_assignment).to_owned(),
1095 is_classic: (self.is_classic),
1096 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
1097 }
1098 }
1099 fn encode_field_0<B: BufMut>(&self, buf: &mut B, version: i16, flex: bool) {
1100 if version >= 0 {
1101 if flex {
1102 let () = put_compact_string(buf, self.member_id);
1103 } else {
1104 let () = put_string(buf, self.member_id);
1105 }
1106 }
1107 }
1108 fn encode_field_1<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
1109 if version >= 0 {
1110 put_i32(buf, self.member_epoch);
1111 }
1112 }
1113 fn encode_field_2<B: BufMut>(&self, buf: &mut B, version: i16, flex: bool) {
1114 if version >= 0 {
1115 if flex {
1116 let () = put_compact_nullable_string(buf, self.instance_id);
1117 } else {
1118 let () = put_nullable_string(buf, self.instance_id);
1119 }
1120 }
1121 }
1122 fn encode_field_3<B: BufMut>(&self, buf: &mut B, version: i16, flex: bool) {
1123 if version >= 0 {
1124 if flex {
1125 let () = put_compact_nullable_string(buf, self.rack_id);
1126 } else {
1127 let () = put_nullable_string(buf, self.rack_id);
1128 }
1129 }
1130 }
1131 fn encode_field_4<B: BufMut>(&self, buf: &mut B, version: i16, flex: bool) {
1132 if version >= 0 {
1133 if flex {
1134 let () = put_compact_string(buf, self.client_id);
1135 } else {
1136 let () = put_string(buf, self.client_id);
1137 }
1138 }
1139 }
1140 fn encode_field_5<B: BufMut>(&self, buf: &mut B, version: i16, flex: bool) {
1141 if version >= 0 {
1142 if flex {
1143 let () = put_compact_string(buf, self.client_host);
1144 } else {
1145 let () = put_string(buf, self.client_host);
1146 }
1147 }
1148 }
1149 fn encode_field_6<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
1150 if version >= 0 {
1151 put_i32(buf, self.topology_epoch);
1152 }
1153 }
1154 fn encode_field_7<B: BufMut>(&self, buf: &mut B, version: i16, flex: bool) {
1155 if version >= 0 {
1156 if flex {
1157 let () = put_compact_string(buf, self.process_id);
1158 } else {
1159 let () = put_string(buf, self.process_id);
1160 }
1161 }
1162 }
1163 fn encode_field_8<B: BufMut>(
1164 &self,
1165 buf: &mut B,
1166 version: i16,
1167 _flex: bool,
1168 ) -> Result<(), ProtocolError> {
1169 if version >= 0 {
1170 match &self.user_endpoint {
1171 None => {
1172 buf.put_i8(-1);
1173 }
1174 Some(v) => {
1175 buf.put_i8(1);
1176 v.encode(buf, version)?;
1177 }
1178 }
1179 }
1180 Ok(())
1181 }
1182 fn encode_field_9<B: BufMut>(
1183 &self,
1184 buf: &mut B,
1185 version: i16,
1186 flex: bool,
1187 ) -> Result<(), ProtocolError> {
1188 if version >= 0 {
1189 {
1190 crate::primitives::array::put_array_len(buf, (self.client_tags).len(), flex);
1191 for it in &self.client_tags {
1192 it.encode(buf, version)?;
1193 }
1194 };
1195 }
1196 Ok(())
1197 }
1198 fn encode_field_10<B: BufMut>(
1199 &self,
1200 buf: &mut B,
1201 version: i16,
1202 flex: bool,
1203 ) -> Result<(), ProtocolError> {
1204 if version >= 0 {
1205 {
1206 crate::primitives::array::put_array_len(buf, (self.task_offsets).len(), flex);
1207 for it in &self.task_offsets {
1208 it.encode(buf, version)?;
1209 }
1210 };
1211 }
1212 Ok(())
1213 }
1214 fn encode_field_11<B: BufMut>(
1215 &self,
1216 buf: &mut B,
1217 version: i16,
1218 flex: bool,
1219 ) -> Result<(), ProtocolError> {
1220 if version >= 0 {
1221 {
1222 crate::primitives::array::put_array_len(buf, (self.task_end_offsets).len(), flex);
1223 for it in &self.task_end_offsets {
1224 it.encode(buf, version)?;
1225 }
1226 };
1227 }
1228 Ok(())
1229 }
1230 fn encode_field_12<B: BufMut>(
1231 &self,
1232 buf: &mut B,
1233 version: i16,
1234 _flex: bool,
1235 ) -> Result<(), ProtocolError> {
1236 if version >= 0 {
1237 self.assignment.encode(buf, version)?;
1238 }
1239 Ok(())
1240 }
1241 fn encode_field_13<B: BufMut>(
1242 &self,
1243 buf: &mut B,
1244 version: i16,
1245 _flex: bool,
1246 ) -> Result<(), ProtocolError> {
1247 if version >= 0 {
1248 self.target_assignment.encode(buf, version)?;
1249 }
1250 Ok(())
1251 }
1252 fn encode_field_14<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
1253 if version >= 0 {
1254 put_bool(buf, self.is_classic);
1255 }
1256 }
1257 fn encode_tagged_fields<B: BufMut>(&self, buf: &mut B, _version: i16, flex: bool) {
1258 if flex {
1259 let tagged = WriteTaggedFields::new();
1260 tagged.write(buf, &self.unknown_tagged_fields);
1261 }
1262 }
1263 fn decode_field_0(
1264 out: &mut Self,
1265 buf: &mut &'a [u8],
1266 version: i16,
1267 flex: bool,
1268 ) -> Result<(), ProtocolError> {
1269 if version >= 0 {
1270 out.member_id = if flex {
1271 get_compact_string_borrowed(buf)?
1272 } else {
1273 get_string_borrowed(buf)?
1274 };
1275 }
1276 Ok(())
1277 }
1278 fn decode_field_1(
1279 out: &mut Self,
1280 buf: &mut &'a [u8],
1281 version: i16,
1282 _flex: bool,
1283 ) -> Result<(), ProtocolError> {
1284 if version >= 0 {
1285 out.member_epoch = get_i32(buf)?;
1286 }
1287 Ok(())
1288 }
1289 fn decode_field_2(
1290 out: &mut Self,
1291 buf: &mut &'a [u8],
1292 version: i16,
1293 flex: bool,
1294 ) -> Result<(), ProtocolError> {
1295 if version >= 0 {
1296 out.instance_id = if flex {
1297 get_compact_nullable_string_borrowed(buf)?
1298 } else {
1299 get_nullable_string_borrowed(buf)?
1300 };
1301 }
1302 Ok(())
1303 }
1304 fn decode_field_3(
1305 out: &mut Self,
1306 buf: &mut &'a [u8],
1307 version: i16,
1308 flex: bool,
1309 ) -> Result<(), ProtocolError> {
1310 if version >= 0 {
1311 out.rack_id = if flex {
1312 get_compact_nullable_string_borrowed(buf)?
1313 } else {
1314 get_nullable_string_borrowed(buf)?
1315 };
1316 }
1317 Ok(())
1318 }
1319 fn decode_field_4(
1320 out: &mut Self,
1321 buf: &mut &'a [u8],
1322 version: i16,
1323 flex: bool,
1324 ) -> Result<(), ProtocolError> {
1325 if version >= 0 {
1326 out.client_id = if flex {
1327 get_compact_string_borrowed(buf)?
1328 } else {
1329 get_string_borrowed(buf)?
1330 };
1331 }
1332 Ok(())
1333 }
1334 fn decode_field_5(
1335 out: &mut Self,
1336 buf: &mut &'a [u8],
1337 version: i16,
1338 flex: bool,
1339 ) -> Result<(), ProtocolError> {
1340 if version >= 0 {
1341 out.client_host = if flex {
1342 get_compact_string_borrowed(buf)?
1343 } else {
1344 get_string_borrowed(buf)?
1345 };
1346 }
1347 Ok(())
1348 }
1349 fn decode_field_6(
1350 out: &mut Self,
1351 buf: &mut &'a [u8],
1352 version: i16,
1353 _flex: bool,
1354 ) -> Result<(), ProtocolError> {
1355 if version >= 0 {
1356 out.topology_epoch = get_i32(buf)?;
1357 }
1358 Ok(())
1359 }
1360 fn decode_field_7(
1361 out: &mut Self,
1362 buf: &mut &'a [u8],
1363 version: i16,
1364 flex: bool,
1365 ) -> Result<(), ProtocolError> {
1366 if version >= 0 {
1367 out.process_id = if flex {
1368 get_compact_string_borrowed(buf)?
1369 } else {
1370 get_string_borrowed(buf)?
1371 };
1372 }
1373 Ok(())
1374 }
1375 fn decode_field_8(
1376 out: &mut Self,
1377 buf: &mut &'a [u8],
1378 version: i16,
1379 _flex: bool,
1380 ) -> Result<(), ProtocolError> {
1381 if version >= 0 {
1382 out.user_endpoint = if get_i8(buf)? < 0 {
1383 None
1384 } else {
1385 Some(super::common::streams_group_describe_response::endpoint::Endpoint::decode_borrow(buf, version)?)
1386 };
1387 }
1388 Ok(())
1389 }
1390 fn decode_field_9(
1391 out: &mut Self,
1392 buf: &mut &'a [u8],
1393 version: i16,
1394 flex: bool,
1395 ) -> Result<(), ProtocolError> {
1396 if version >= 0 {
1397 out.client_tags = {
1398 let n = crate::primitives::array::get_array_len(buf, flex)?;
1399 let mut v = Vec::with_capacity(n);
1400 for _ in 0..n {
1401 v.push(super::common::streams_group_describe_response::key_value::KeyValue::decode_borrow(buf, version)?);
1402 }
1403 v
1404 };
1405 }
1406 Ok(())
1407 }
1408 fn decode_field_10(
1409 out: &mut Self,
1410 buf: &mut &'a [u8],
1411 version: i16,
1412 flex: bool,
1413 ) -> Result<(), ProtocolError> {
1414 if version >= 0 {
1415 out.task_offsets = {
1416 let n = crate::primitives::array::get_array_len(buf, flex)?;
1417 let mut v = Vec::with_capacity(n);
1418 for _ in 0..n {
1419 v.push(super::common::streams_group_describe_response::task_offset::TaskOffset::decode_borrow(buf, version)?);
1420 }
1421 v
1422 };
1423 }
1424 Ok(())
1425 }
1426 fn decode_field_11(
1427 out: &mut Self,
1428 buf: &mut &'a [u8],
1429 version: i16,
1430 flex: bool,
1431 ) -> Result<(), ProtocolError> {
1432 if version >= 0 {
1433 out.task_end_offsets = {
1434 let n = crate::primitives::array::get_array_len(buf, flex)?;
1435 let mut v = Vec::with_capacity(n);
1436 for _ in 0..n {
1437 v.push(super::common::streams_group_describe_response::task_offset::TaskOffset::decode_borrow(buf, version)?);
1438 }
1439 v
1440 };
1441 }
1442 Ok(())
1443 }
1444 fn decode_field_12(
1445 out: &mut Self,
1446 buf: &mut &'a [u8],
1447 version: i16,
1448 _flex: bool,
1449 ) -> Result<(), ProtocolError> {
1450 if version >= 0 {
1451 out.assignment = super::common::streams_group_describe_response::assignment::Assignment::decode_borrow(buf, version)?;
1452 }
1453 Ok(())
1454 }
1455 fn decode_field_13(
1456 out: &mut Self,
1457 buf: &mut &'a [u8],
1458 version: i16,
1459 _flex: bool,
1460 ) -> Result<(), ProtocolError> {
1461 if version >= 0 {
1462 out.target_assignment = super::common::streams_group_describe_response::assignment::Assignment::decode_borrow(buf, version)?;
1463 }
1464 Ok(())
1465 }
1466 fn decode_field_14(
1467 out: &mut Self,
1468 buf: &mut &'a [u8],
1469 version: i16,
1470 _flex: bool,
1471 ) -> Result<(), ProtocolError> {
1472 if version >= 0 {
1473 out.is_classic = get_bool(buf)?;
1474 }
1475 Ok(())
1476 }
1477 fn decode_tagged_fields(
1478 out: &mut Self,
1479 buf: &mut &'a [u8],
1480 _version: i16,
1481 flex: bool,
1482 ) -> Result<(), ProtocolError> {
1483 if flex {
1484 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
1485 }
1486 Ok(())
1487 }
1488}
1489impl Encode for Member<'_> {
1490 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
1491 let flex = version >= 0;
1492 self.encode_field_0(buf, version, flex);
1493 self.encode_field_1(buf, version, flex);
1494 self.encode_field_2(buf, version, flex);
1495 self.encode_field_3(buf, version, flex);
1496 self.encode_field_4(buf, version, flex);
1497 self.encode_field_5(buf, version, flex);
1498 self.encode_field_6(buf, version, flex);
1499 self.encode_field_7(buf, version, flex);
1500 self.encode_field_8(buf, version, flex)?;
1501 self.encode_field_9(buf, version, flex)?;
1502 self.encode_field_10(buf, version, flex)?;
1503 self.encode_field_11(buf, version, flex)?;
1504 self.encode_field_12(buf, version, flex)?;
1505 self.encode_field_13(buf, version, flex)?;
1506 self.encode_field_14(buf, version, flex);
1507 self.encode_tagged_fields(buf, version, flex);
1508 Ok(())
1509 }
1510 fn encoded_len(&self, version: i16) -> usize {
1511 let flex = version >= 0;
1512 let mut n: usize = 0;
1513 if version >= 0 {
1514 n += if flex {
1515 compact_string_len(self.member_id)
1516 } else {
1517 string_len(self.member_id)
1518 };
1519 }
1520 if version >= 0 {
1521 n += 4;
1522 }
1523 if version >= 0 {
1524 n += if flex {
1525 compact_nullable_string_len(self.instance_id)
1526 } else {
1527 nullable_string_len(self.instance_id)
1528 };
1529 }
1530 if version >= 0 {
1531 n += if flex {
1532 compact_nullable_string_len(self.rack_id)
1533 } else {
1534 nullable_string_len(self.rack_id)
1535 };
1536 }
1537 if version >= 0 {
1538 n += if flex {
1539 compact_string_len(self.client_id)
1540 } else {
1541 string_len(self.client_id)
1542 };
1543 }
1544 if version >= 0 {
1545 n += if flex {
1546 compact_string_len(self.client_host)
1547 } else {
1548 string_len(self.client_host)
1549 };
1550 }
1551 if version >= 0 {
1552 n += 4;
1553 }
1554 if version >= 0 {
1555 n += if flex {
1556 compact_string_len(self.process_id)
1557 } else {
1558 string_len(self.process_id)
1559 };
1560 }
1561 if version >= 0 {
1562 n += 1 + self
1563 .user_endpoint
1564 .as_ref()
1565 .map_or(0, |v| v.encoded_len(version));
1566 }
1567 if version >= 0 {
1568 n += {
1569 let prefix =
1570 crate::primitives::array::array_len_prefix_len((self.client_tags).len(), flex);
1571 let body: usize = (self.client_tags)
1572 .iter()
1573 .map(|it| it.encoded_len(version))
1574 .sum();
1575 prefix + body
1576 };
1577 }
1578 if version >= 0 {
1579 n += {
1580 let prefix =
1581 crate::primitives::array::array_len_prefix_len((self.task_offsets).len(), flex);
1582 let body: usize = (self.task_offsets)
1583 .iter()
1584 .map(|it| it.encoded_len(version))
1585 .sum();
1586 prefix + body
1587 };
1588 }
1589 if version >= 0 {
1590 n += {
1591 let prefix = crate::primitives::array::array_len_prefix_len(
1592 (self.task_end_offsets).len(),
1593 flex,
1594 );
1595 let body: usize = (self.task_end_offsets)
1596 .iter()
1597 .map(|it| it.encoded_len(version))
1598 .sum();
1599 prefix + body
1600 };
1601 }
1602 if version >= 0 {
1603 n += self.assignment.encoded_len(version);
1604 }
1605 if version >= 0 {
1606 n += self.target_assignment.encoded_len(version);
1607 }
1608 if version >= 0 {
1609 n += 1;
1610 }
1611 if flex {
1612 let known_pairs: Vec<(u32, usize)> = Vec::new();
1613 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
1614 }
1615 n
1616 }
1617}
1618impl<'de> DecodeBorrow<'de> for Member<'de> {
1619 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
1620 let flex = version >= 0;
1621 let mut out = Self::default();
1622 Self::decode_field_0(&mut out, buf, version, flex)?;
1623 Self::decode_field_1(&mut out, buf, version, flex)?;
1624 Self::decode_field_2(&mut out, buf, version, flex)?;
1625 Self::decode_field_3(&mut out, buf, version, flex)?;
1626 Self::decode_field_4(&mut out, buf, version, flex)?;
1627 Self::decode_field_5(&mut out, buf, version, flex)?;
1628 Self::decode_field_6(&mut out, buf, version, flex)?;
1629 Self::decode_field_7(&mut out, buf, version, flex)?;
1630 Self::decode_field_8(&mut out, buf, version, flex)?;
1631 Self::decode_field_9(&mut out, buf, version, flex)?;
1632 Self::decode_field_10(&mut out, buf, version, flex)?;
1633 Self::decode_field_11(&mut out, buf, version, flex)?;
1634 Self::decode_field_12(&mut out, buf, version, flex)?;
1635 Self::decode_field_13(&mut out, buf, version, flex)?;
1636 Self::decode_field_14(&mut out, buf, version, flex)?;
1637 Self::decode_tagged_fields(&mut out, buf, version, flex)?;
1638 Ok(out)
1639 }
1640}
1641#[cfg(test)]
1642impl Member<'_> {
1643 #[must_use]
1644 pub fn populated(version: i16) -> Self {
1645 let mut m = Self::default();
1646 if version >= 0 {
1647 m.member_id = "x";
1648 }
1649 if version >= 0 {
1650 m.member_epoch = 1i32;
1651 }
1652 if version >= 0 {
1653 m.instance_id = Some("x");
1654 }
1655 if version >= 0 {
1656 m.rack_id = Some("x");
1657 }
1658 if version >= 0 {
1659 m.client_id = "x";
1660 }
1661 if version >= 0 {
1662 m.client_host = "x";
1663 }
1664 if version >= 0 {
1665 m.topology_epoch = 1i32;
1666 }
1667 if version >= 0 {
1668 m.process_id = "x";
1669 }
1670 if version >= 0 {
1671 m.user_endpoint = Some(
1672 super::common::streams_group_describe_response::endpoint::Endpoint::populated(
1673 version,
1674 ),
1675 );
1676 }
1677 if version >= 0 {
1678 m.client_tags = vec![
1679 super::common::streams_group_describe_response::key_value::KeyValue::populated(
1680 version,
1681 ),
1682 ];
1683 }
1684 if version >= 0 {
1685 m.task_offsets = vec![
1686 super::common::streams_group_describe_response::task_offset::TaskOffset::populated(
1687 version,
1688 ),
1689 ];
1690 }
1691 if version >= 0 {
1692 m.task_end_offsets = vec![
1693 super::common::streams_group_describe_response::task_offset::TaskOffset::populated(
1694 version,
1695 ),
1696 ];
1697 }
1698 if version >= 0 {
1699 m.assignment =
1700 super::common::streams_group_describe_response::assignment::Assignment::populated(
1701 version,
1702 );
1703 }
1704 if version >= 0 {
1705 m.target_assignment =
1706 super::common::streams_group_describe_response::assignment::Assignment::populated(
1707 version,
1708 );
1709 }
1710 if version >= 0 {
1711 m.is_classic = true;
1712 }
1713 m
1714 }
1715}