krabka_protocol/opt/rustwide/workdir/generated/
StreamsGroupHeartbeatRequest.owned.rs

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

krabka_protocol/opt/rustwide/workdir/generated/
StreamsGroupHeartbeatRequest.owned.rs
