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