krabka_protocol/opt/rustwide/workdir/generated/
FetchResponse.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    codegen_helpers::is_default,
8    primitives::{
9        array::{
10            array_len_prefix_len, get_array_len, get_nullable_array_len,
11            nullable_array_len_prefix_len, put_array_len, put_nullable_array_len,
12        },
13        fixed::{get_i16, get_i32, get_i64, put_i16, put_i32, put_i64},
14        string_bytes::{
15            compact_bytes_len_from_size, compact_nullable_string_len, compact_string_len,
16            get_compact_nullable_bytes_owned, get_compact_nullable_string_owned,
17            get_compact_string_owned, get_nullable_bytes_owned, get_nullable_string_owned,
18            get_string_owned, nullable_string_len, put_compact_nullable_bytes,
19            put_compact_nullable_string, put_compact_string, put_nullable_bytes,
20            put_nullable_string, put_string, string_len,
21        },
22        uuid::{Uuid, get_uuid, put_uuid},
23        varint::uvarint_len,
24    },
25    records::RecordsPayload,
26    tagged_fields::{WriteTaggedFields, encode_to_bytes, read_tagged_fields, tagged_fields_len},
27};
28pub const API_KEY: i16 = 1;
29pub const MIN_VERSION: i16 = 4;
30pub const MAX_VERSION: i16 = 18;
31pub const FLEXIBLE_MIN: i16 = 12;
32#[inline]
33#[must_use]
34pub fn is_flexible(version: i16) -> bool {
35    version >= FLEXIBLE_MIN
36}
37#[derive(Debug, Clone, PartialEq, Eq, Default)]
38pub struct FetchResponse {
39    pub throttle_time_ms: i32,
40    pub error_code: i16,
41    pub session_id: i32,
42    pub responses: Vec<FetchableTopicResponse>,
43    pub node_endpoints: Vec<NodeEndpoint>,
44    pub unknown_tagged_fields: UnknownTaggedFields,
45}
46impl Encode for FetchResponse {
47    fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
48        if !(MIN_VERSION..=MAX_VERSION).contains(&version) {
49            return Err(ProtocolError::UnsupportedVersion {
50                api_key: API_KEY,
51                version,
52            });
53        }
54        let flex = is_flexible(version);
55        if version >= 1 {
56            put_i32(buf, self.throttle_time_ms);
57        }
58        if version >= 7 {
59            put_i16(buf, self.error_code);
60        }
61        if version >= 7 {
62            put_i32(buf, self.session_id);
63        }
64        if version >= 0 {
65            {
66                put_array_len(buf, (self.responses).len(), flex)?;
67                for it in &self.responses {
68                    it.encode(buf, version)?;
69                }
70            }
71        }
72        if flex {
73            let mut tagged = WriteTaggedFields::new();
74            if version >= 16 && !(is_default(&self.node_endpoints)) {
75                let payload = encode_to_bytes(
76                    {
77                        let prefix = array_len_prefix_len((self.node_endpoints).len(), flex);
78                        let body: usize = (self.node_endpoints)
79                            .iter()
80                            .map(|it| it.encoded_len(version))
81                            .sum();
82                        prefix + body
83                    },
84                    |b| {
85                        {
86                            put_array_len(b, (self.node_endpoints).len(), flex)?;
87                            for it in &self.node_endpoints {
88                                it.encode(b, version)?;
89                            }
90                        }
91                        Ok(())
92                    },
93                )?;
94                tagged.add(0, payload);
95            }
96            tagged.write(buf, &self.unknown_tagged_fields)?;
97        }
98        Ok(())
99    }
100    fn encoded_len(&self, version: i16) -> usize {
101        let flex = is_flexible(version);
102        let mut n: usize = 0;
103        if version >= 1 {
104            n += 4;
105        }
106        if version >= 7 {
107            n += 2;
108        }
109        if version >= 7 {
110            n += 4;
111        }
112        if version >= 0 {
113            n += {
114                let prefix = array_len_prefix_len((self.responses).len(), flex);
115                let body: usize = (self.responses)
116                    .iter()
117                    .map(|it| it.encoded_len(version))
118                    .sum();
119                prefix + body
120            };
121        }
122        if flex {
123            let mut known_pairs: Vec<(u32, usize)> = Vec::new();
124            if version >= 16 && !(is_default(&self.node_endpoints)) {
125                known_pairs.push((0, {
126                    let prefix = array_len_prefix_len((self.node_endpoints).len(), flex);
127                    let body: usize = (self.node_endpoints)
128                        .iter()
129                        .map(|it| it.encoded_len(version))
130                        .sum();
131                    prefix + body
132                }));
133            }
134            n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
135        }
136        n
137    }
138}
139impl Decode<'_> for FetchResponse {
140    fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
141        if !(MIN_VERSION..=MAX_VERSION).contains(&version) {
142            return Err(ProtocolError::UnsupportedVersion {
143                api_key: API_KEY,
144                version,
145            });
146        }
147        let flex = is_flexible(version);
148        let mut out = Self::default();
149        if version >= 1 {
150            out.throttle_time_ms = get_i32(buf)?;
151        }
152        if version >= 7 {
153            out.error_code = get_i16(buf)?;
154        }
155        if version >= 7 {
156            out.session_id = get_i32(buf)?;
157        }
158        if version >= 0 {
159            out.responses = {
160                let n = get_array_len(buf, flex)?;
161                let mut v = Vec::with_capacity(n);
162                for _ in 0..n {
163                    v.push(FetchableTopicResponse::decode(buf, version)?);
164                }
165                v
166            };
167        }
168        if flex {
169            let mut tag_node_endpoints = None;
170            out.unknown_tagged_fields = read_tagged_fields(buf, |tag, payload| match tag {
171                0 if version >= 16 => {
172                    tag_node_endpoints = Some({
173                        let b: &mut &[u8] = payload;
174                        {
175                            let n = get_array_len(b, flex)?;
176                            let mut v = Vec::with_capacity(n);
177                            for _ in 0..n {
178                                v.push(NodeEndpoint::decode(b, version)?);
179                            }
180                            v
181                        }
182                    });
183                    Ok(true)
184                }
185                0 => Err(ProtocolError::TagNotValidForVersion { tag: 0, version }),
186                _ => Ok(false),
187            })?;
188            if let Some(v) = tag_node_endpoints {
189                out.node_endpoints = v;
190            }
191        }
192        Ok(out)
193    }
194}
195#[cfg(test)]
196impl FetchResponse {
197    #[must_use]
198    pub fn populated(version: i16) -> Self {
199        let mut m = Self::default();
200        if version >= 1 {
201            m.throttle_time_ms = 1i32;
202        }
203        if version >= 7 {
204            m.error_code = 1i16;
205        }
206        if version >= 7 {
207            m.session_id = 1i32;
208        }
209        if version >= 0 {
210            m.responses = vec![FetchableTopicResponse::populated(version)];
211        }
212        if version >= 16 {
213            m.node_endpoints = vec![NodeEndpoint::populated(version)];
214        }
215        m
216    }
217}
218#[derive(Debug, Clone, PartialEq, Eq, Default)]
219pub struct FetchableTopicResponse {
220    pub topic: String,
221    pub topic_id: Uuid,
222    pub partitions: Vec<PartitionData>,
223    pub unknown_tagged_fields: UnknownTaggedFields,
224}
225impl Encode for FetchableTopicResponse {
226    fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
227        let flex = version >= 12;
228        if (0..=12).contains(&version) {
229            if flex {
230                put_compact_string(buf, &self.topic)?;
231            } else {
232                put_string(buf, &self.topic)?;
233            }
234        }
235        if version >= 13 {
236            put_uuid(buf, self.topic_id);
237        }
238        if version >= 0 {
239            {
240                put_array_len(buf, (self.partitions).len(), flex)?;
241                for it in &self.partitions {
242                    it.encode(buf, version)?;
243                }
244            }
245        }
246        if flex {
247            let tagged = WriteTaggedFields::new();
248            tagged.write(buf, &self.unknown_tagged_fields)?;
249        }
250        Ok(())
251    }
252    fn encoded_len(&self, version: i16) -> usize {
253        let flex = version >= 12;
254        let mut n: usize = 0;
255        if (0..=12).contains(&version) {
256            n += if flex {
257                compact_string_len(&self.topic)
258            } else {
259                string_len(&self.topic)
260            };
261        }
262        if version >= 13 {
263            n += 16;
264        }
265        if version >= 0 {
266            n += {
267                let prefix = array_len_prefix_len((self.partitions).len(), flex);
268                let body: usize = (self.partitions)
269                    .iter()
270                    .map(|it| it.encoded_len(version))
271                    .sum();
272                prefix + body
273            };
274        }
275        if flex {
276            let known_pairs: Vec<(u32, usize)> = Vec::new();
277            n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
278        }
279        n
280    }
281}
282impl Decode<'_> for FetchableTopicResponse {
283    fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
284        let flex = version >= 12;
285        let mut out = Self::default();
286        if (0..=12).contains(&version) {
287            out.topic = if flex {
288                get_compact_string_owned(buf)?
289            } else {
290                get_string_owned(buf)?
291            };
292        }
293        if version >= 13 {
294            out.topic_id = get_uuid(buf)?;
295        }
296        if version >= 0 {
297            out.partitions = {
298                let n = get_array_len(buf, flex)?;
299                let mut v = Vec::with_capacity(n);
300                for _ in 0..n {
301                    v.push(PartitionData::decode(buf, version)?);
302                }
303                v
304            };
305        }
306        if flex {
307            out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
308        }
309        Ok(out)
310    }
311}
312#[cfg(test)]
313impl FetchableTopicResponse {
314    #[must_use]
315    pub fn populated(version: i16) -> Self {
316        let mut m = Self::default();
317        if (0..=12).contains(&version) {
318            m.topic = "x".to_owned();
319        }
320        if version >= 13 {
321            m.topic_id = Uuid([1u8; 16]);
322        }
323        if version >= 0 {
324            m.partitions = vec![PartitionData::populated(version)];
325        }
326        m
327    }
328}
329#[derive(Debug, Clone, PartialEq, Eq)]
330pub struct PartitionData {
331    pub partition_index: i32,
332    pub error_code: i16,
333    pub high_watermark: i64,
334    pub last_stable_offset: i64,
335    pub log_start_offset: i64,
336    pub aborted_transactions: Option<Vec<AbortedTransaction>>,
337    pub preferred_read_replica: i32,
338    pub records: Option<RecordsPayload>,
339    pub diverging_epoch: EpochEndOffset,
340    pub current_leader: LeaderIdAndEpoch,
341    pub snapshot_id: SnapshotId,
342    pub unknown_tagged_fields: UnknownTaggedFields,
343}
344impl Default for PartitionData {
345    fn default() -> Self {
346        Self {
347            partition_index: 0i32,
348            error_code: 0i16,
349            high_watermark: 0i64,
350            last_stable_offset: -1i64,
351            log_start_offset: -1i64,
352            aborted_transactions: Some(Vec::new()),
353            preferred_read_replica: -1i32,
354            records: None,
355            diverging_epoch: EpochEndOffset::default(),
356            current_leader: LeaderIdAndEpoch::default(),
357            snapshot_id: SnapshotId::default(),
358            unknown_tagged_fields: UnknownTaggedFields::default(),
359        }
360    }
361}
362impl PartitionData {
363    fn encode_field_0<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
364        if version >= 0 {
365            put_i32(buf, self.partition_index);
366        }
367    }
368    fn encode_field_1<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
369        if version >= 0 {
370            put_i16(buf, self.error_code);
371        }
372    }
373    fn encode_field_2<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
374        if version >= 0 {
375            put_i64(buf, self.high_watermark);
376        }
377    }
378    fn encode_field_3<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
379        if version >= 4 {
380            put_i64(buf, self.last_stable_offset);
381        }
382    }
383    fn encode_field_4<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
384        if version >= 5 {
385            put_i64(buf, self.log_start_offset);
386        }
387    }
388    fn encode_field_5<B: BufMut>(
389        &self,
390        buf: &mut B,
391        version: i16,
392        flex: bool,
393    ) -> Result<(), ProtocolError> {
394        if version >= 4 {
395            {
396                let len = (self.aborted_transactions).as_ref().map(Vec::len);
397                put_nullable_array_len(buf, len, flex)?;
398                if let Some(v) = &self.aborted_transactions {
399                    for it in v {
400                        it.encode(buf, version)?;
401                    }
402                }
403            }
404        }
405        Ok(())
406    }
407    fn encode_field_6<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
408        if version >= 11 {
409            put_i32(buf, self.preferred_read_replica);
410        }
411    }
412    fn encode_field_7<B: BufMut>(
413        &self,
414        buf: &mut B,
415        version: i16,
416        flex: bool,
417    ) -> Result<(), ProtocolError> {
418        if version >= 0 {
419            match &self.records {
420                None => {
421                    if flex {
422                        put_compact_nullable_bytes(buf, None)?;
423                    } else {
424                        put_nullable_bytes(buf, None)?;
425                    }
426                }
427                Some(__rb) => __rb.encode_length_prefixed(buf, version, flex)?,
428            }
429        }
430        Ok(())
431    }
432    fn encode_tagged_fields<B: BufMut>(
433        &self,
434        buf: &mut B,
435        version: i16,
436        flex: bool,
437    ) -> Result<(), ProtocolError> {
438        if flex {
439            let mut tagged = WriteTaggedFields::new();
440            if !(is_default(&self.diverging_epoch)) {
441                let payload = encode_to_bytes(self.diverging_epoch.encoded_len(version), |b| {
442                    self.diverging_epoch.encode(b, version)?;
443                    Ok(())
444                })?;
445                tagged.add(0, payload);
446            }
447            if !(is_default(&self.current_leader)) {
448                let payload = encode_to_bytes(self.current_leader.encoded_len(version), |b| {
449                    self.current_leader.encode(b, version)?;
450                    Ok(())
451                })?;
452                tagged.add(1, payload);
453            }
454            if !(is_default(&self.snapshot_id)) {
455                let payload = encode_to_bytes(self.snapshot_id.encoded_len(version), |b| {
456                    self.snapshot_id.encode(b, version)?;
457                    Ok(())
458                })?;
459                tagged.add(2, payload);
460            }
461            tagged.write(buf, &self.unknown_tagged_fields)?;
462        }
463        Ok(())
464    }
465    fn decode_field_0<B: Buf>(
466        out: &mut Self,
467        buf: &mut B,
468        version: i16,
469        _flex: bool,
470    ) -> Result<(), ProtocolError> {
471        if version >= 0 {
472            out.partition_index = get_i32(buf)?;
473        }
474        Ok(())
475    }
476    fn decode_field_1<B: Buf>(
477        out: &mut Self,
478        buf: &mut B,
479        version: i16,
480        _flex: bool,
481    ) -> Result<(), ProtocolError> {
482        if version >= 0 {
483            out.error_code = get_i16(buf)?;
484        }
485        Ok(())
486    }
487    fn decode_field_2<B: Buf>(
488        out: &mut Self,
489        buf: &mut B,
490        version: i16,
491        _flex: bool,
492    ) -> Result<(), ProtocolError> {
493        if version >= 0 {
494            out.high_watermark = get_i64(buf)?;
495        }
496        Ok(())
497    }
498    fn decode_field_3<B: Buf>(
499        out: &mut Self,
500        buf: &mut B,
501        version: i16,
502        _flex: bool,
503    ) -> Result<(), ProtocolError> {
504        if version >= 4 {
505            out.last_stable_offset = get_i64(buf)?;
506        }
507        Ok(())
508    }
509    fn decode_field_4<B: Buf>(
510        out: &mut Self,
511        buf: &mut B,
512        version: i16,
513        _flex: bool,
514    ) -> Result<(), ProtocolError> {
515        if version >= 5 {
516            out.log_start_offset = get_i64(buf)?;
517        }
518        Ok(())
519    }
520    fn decode_field_5<B: Buf>(
521        out: &mut Self,
522        buf: &mut B,
523        version: i16,
524        flex: bool,
525    ) -> Result<(), ProtocolError> {
526        if version >= 4 {
527            out.aborted_transactions = {
528                let opt = get_nullable_array_len(buf, flex)?;
529                match opt {
530                    None => None,
531                    Some(n) => {
532                        let mut v = Vec::with_capacity(n);
533                        for _ in 0..n {
534                            v.push(AbortedTransaction::decode(buf, version)?);
535                        }
536                        Some(v)
537                    }
538                }
539            };
540        }
541        Ok(())
542    }
543    fn decode_field_6<B: Buf>(
544        out: &mut Self,
545        buf: &mut B,
546        version: i16,
547        _flex: bool,
548    ) -> Result<(), ProtocolError> {
549        if version >= 11 {
550            out.preferred_read_replica = get_i32(buf)?;
551        }
552        Ok(())
553    }
554    fn decode_field_7<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.records = {
562                let __rb_opt = if flex {
563                    get_compact_nullable_bytes_owned(buf)?
564                } else {
565                    get_nullable_bytes_owned(buf)?
566                };
567                match __rb_opt {
568                    None => None,
569                    Some(__rb_bytes) => Some(RecordsPayload::from_fetch_bytes(__rb_bytes)?),
570                }
571            };
572        }
573        Ok(())
574    }
575    fn decode_tagged_fields<B: Buf>(
576        out: &mut Self,
577        buf: &mut B,
578        version: i16,
579        flex: bool,
580    ) -> Result<(), ProtocolError> {
581        if flex {
582            let mut tag_diverging_epoch = None;
583            let mut tag_current_leader = None;
584            let mut tag_snapshot_id = None;
585            out.unknown_tagged_fields = read_tagged_fields(buf, |tag, payload| match tag {
586                0 => {
587                    tag_diverging_epoch = Some({
588                        let b: &mut &[u8] = payload;
589                        EpochEndOffset::decode(b, version)?
590                    });
591                    Ok(true)
592                }
593                1 => {
594                    tag_current_leader = Some({
595                        let b: &mut &[u8] = payload;
596                        LeaderIdAndEpoch::decode(b, version)?
597                    });
598                    Ok(true)
599                }
600                2 => {
601                    tag_snapshot_id = Some({
602                        let b: &mut &[u8] = payload;
603                        SnapshotId::decode(b, version)?
604                    });
605                    Ok(true)
606                }
607                _ => Ok(false),
608            })?;
609            if let Some(v) = tag_diverging_epoch {
610                out.diverging_epoch = v;
611            }
612            if let Some(v) = tag_current_leader {
613                out.current_leader = v;
614            }
615            if let Some(v) = tag_snapshot_id {
616                out.snapshot_id = v;
617            }
618        }
619        Ok(())
620    }
621}
622impl Encode for PartitionData {
623    fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
624        let flex = version >= 12;
625        self.encode_field_0(buf, version, flex);
626        self.encode_field_1(buf, version, flex);
627        self.encode_field_2(buf, version, flex);
628        self.encode_field_3(buf, version, flex);
629        self.encode_field_4(buf, version, flex);
630        self.encode_field_5(buf, version, flex)?;
631        self.encode_field_6(buf, version, flex);
632        self.encode_field_7(buf, version, flex)?;
633        self.encode_tagged_fields(buf, version, flex)?;
634        Ok(())
635    }
636    fn encoded_len(&self, version: i16) -> usize {
637        let flex = version >= 12;
638        let mut n: usize = 0;
639        if version >= 0 {
640            n += 4;
641        }
642        if version >= 0 {
643            n += 2;
644        }
645        if version >= 0 {
646            n += 8;
647        }
648        if version >= 4 {
649            n += 8;
650        }
651        if version >= 5 {
652            n += 8;
653        }
654        if version >= 4 {
655            n += {
656                let opt: Option<&Vec<_>> = (self.aborted_transactions).as_ref();
657                let prefix = nullable_array_len_prefix_len(opt.map(Vec::len), flex);
658                let body: usize =
659                    opt.map_or(0, |v| v.iter().map(|it| it.encoded_len(version)).sum());
660                prefix + body
661            };
662        }
663        if version >= 11 {
664            n += 4;
665        }
666        if version >= 0 {
667            n += (self.records).as_ref().map_or_else(
668                || if flex { uvarint_len(0) } else { 4 },
669                |__rb| {
670                    let __rb_len = <RecordsPayload as crate::Encode>::encoded_len(__rb, version);
671                    if flex {
672                        compact_bytes_len_from_size(__rb_len)
673                    } else {
674                        4 + __rb_len
675                    }
676                },
677            );
678        }
679        if flex {
680            let mut known_pairs: Vec<(u32, usize)> = Vec::new();
681            if !(is_default(&self.diverging_epoch)) {
682                known_pairs.push((0, self.diverging_epoch.encoded_len(version)));
683            }
684            if !(is_default(&self.current_leader)) {
685                known_pairs.push((1, self.current_leader.encoded_len(version)));
686            }
687            if !(is_default(&self.snapshot_id)) {
688                known_pairs.push((2, self.snapshot_id.encoded_len(version)));
689            }
690            n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
691        }
692        n
693    }
694}
695impl Decode<'_> for PartitionData {
696    fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
697        let flex = version >= 12;
698        let mut out = Self::default();
699        Self::decode_field_0(&mut out, buf, version, flex)?;
700        Self::decode_field_1(&mut out, buf, version, flex)?;
701        Self::decode_field_2(&mut out, buf, version, flex)?;
702        Self::decode_field_3(&mut out, buf, version, flex)?;
703        Self::decode_field_4(&mut out, buf, version, flex)?;
704        Self::decode_field_5(&mut out, buf, version, flex)?;
705        Self::decode_field_6(&mut out, buf, version, flex)?;
706        Self::decode_field_7(&mut out, buf, version, flex)?;
707        Self::decode_tagged_fields(&mut out, buf, version, flex)?;
708        Ok(out)
709    }
710}
711#[cfg(test)]
712impl PartitionData {
713    #[must_use]
714    pub fn populated(version: i16) -> Self {
715        let mut m = Self::default();
716        if version >= 0 {
717            m.partition_index = 1i32;
718        }
719        if version >= 0 {
720            m.error_code = 1i16;
721        }
722        if version >= 0 {
723            m.high_watermark = 1i64;
724        }
725        if version >= 4 {
726            m.last_stable_offset = 1i64;
727        }
728        if version >= 5 {
729            m.log_start_offset = 1i64;
730        }
731        if version >= 4 {
732            m.aborted_transactions = Some(vec![AbortedTransaction::populated(version)]);
733        }
734        if version >= 11 {
735            m.preferred_read_replica = 1i32;
736        }
737        if version >= 12 {
738            m.diverging_epoch = EpochEndOffset::populated(version);
739        }
740        if version >= 12 {
741            m.current_leader = LeaderIdAndEpoch::populated(version);
742        }
743        if version >= 12 {
744            m.snapshot_id = SnapshotId::populated(version);
745        }
746        m
747    }
748}
749#[derive(Debug, Clone, PartialEq, Eq)]
750pub struct EpochEndOffset {
751    pub epoch: i32,
752    pub end_offset: i64,
753    pub unknown_tagged_fields: UnknownTaggedFields,
754}
755impl Default for EpochEndOffset {
756    fn default() -> Self {
757        Self {
758            epoch: -1i32,
759            end_offset: -1i64,
760            unknown_tagged_fields: UnknownTaggedFields::default(),
761        }
762    }
763}
764impl Encode for EpochEndOffset {
765    fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
766        let flex = version >= 12;
767        if version >= 12 {
768            put_i32(buf, self.epoch);
769        }
770        if version >= 12 {
771            put_i64(buf, self.end_offset);
772        }
773        if flex {
774            let tagged = WriteTaggedFields::new();
775            tagged.write(buf, &self.unknown_tagged_fields)?;
776        }
777        Ok(())
778    }
779    fn encoded_len(&self, version: i16) -> usize {
780        let flex = version >= 12;
781        let mut n: usize = 0;
782        if version >= 12 {
783            n += 4;
784        }
785        if version >= 12 {
786            n += 8;
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 EpochEndOffset {
796    fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
797        let flex = version >= 12;
798        let mut out = Self::default();
799        if version >= 12 {
800            out.epoch = get_i32(buf)?;
801        }
802        if version >= 12 {
803            out.end_offset = get_i64(buf)?;
804        }
805        if flex {
806            out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
807        }
808        Ok(out)
809    }
810}
811#[cfg(test)]
812impl EpochEndOffset {
813    #[must_use]
814    pub fn populated(version: i16) -> Self {
815        let mut m = Self::default();
816        if version >= 12 {
817            m.epoch = 1i32;
818        }
819        if version >= 12 {
820            m.end_offset = 1i64;
821        }
822        m
823    }
824}
825#[derive(Debug, Clone, PartialEq, Eq)]
826pub struct LeaderIdAndEpoch {
827    pub leader_id: i32,
828    pub leader_epoch: i32,
829    pub unknown_tagged_fields: UnknownTaggedFields,
830}
831impl Default for LeaderIdAndEpoch {
832    fn default() -> Self {
833        Self {
834            leader_id: -1i32,
835            leader_epoch: -1i32,
836            unknown_tagged_fields: UnknownTaggedFields::default(),
837        }
838    }
839}
840impl Encode for LeaderIdAndEpoch {
841    fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
842        let flex = version >= 12;
843        if version >= 12 {
844            put_i32(buf, self.leader_id);
845        }
846        if version >= 12 {
847            put_i32(buf, self.leader_epoch);
848        }
849        if flex {
850            let tagged = WriteTaggedFields::new();
851            tagged.write(buf, &self.unknown_tagged_fields)?;
852        }
853        Ok(())
854    }
855    fn encoded_len(&self, version: i16) -> usize {
856        let flex = version >= 12;
857        let mut n: usize = 0;
858        if version >= 12 {
859            n += 4;
860        }
861        if version >= 12 {
862            n += 4;
863        }
864        if flex {
865            let known_pairs: Vec<(u32, usize)> = Vec::new();
866            n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
867        }
868        n
869    }
870}
871impl Decode<'_> for LeaderIdAndEpoch {
872    fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
873        let flex = version >= 12;
874        let mut out = Self::default();
875        if version >= 12 {
876            out.leader_id = get_i32(buf)?;
877        }
878        if version >= 12 {
879            out.leader_epoch = get_i32(buf)?;
880        }
881        if flex {
882            out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
883        }
884        Ok(out)
885    }
886}
887#[cfg(test)]
888impl LeaderIdAndEpoch {
889    #[must_use]
890    pub fn populated(version: i16) -> Self {
891        let mut m = Self::default();
892        if version >= 12 {
893            m.leader_id = 1i32;
894        }
895        if version >= 12 {
896            m.leader_epoch = 1i32;
897        }
898        m
899    }
900}
901#[derive(Debug, Clone, PartialEq, Eq)]
902pub struct SnapshotId {
903    pub end_offset: i64,
904    pub epoch: i32,
905    pub unknown_tagged_fields: UnknownTaggedFields,
906}
907impl Default for SnapshotId {
908    fn default() -> Self {
909        Self {
910            end_offset: -1i64,
911            epoch: -1i32,
912            unknown_tagged_fields: UnknownTaggedFields::default(),
913        }
914    }
915}
916impl Encode for SnapshotId {
917    fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
918        let flex = version >= 12;
919        if version >= 0 {
920            put_i64(buf, self.end_offset);
921        }
922        if version >= 0 {
923            put_i32(buf, self.epoch);
924        }
925        if flex {
926            let tagged = WriteTaggedFields::new();
927            tagged.write(buf, &self.unknown_tagged_fields)?;
928        }
929        Ok(())
930    }
931    fn encoded_len(&self, version: i16) -> usize {
932        let flex = version >= 12;
933        let mut n: usize = 0;
934        if version >= 0 {
935            n += 8;
936        }
937        if version >= 0 {
938            n += 4;
939        }
940        if flex {
941            let known_pairs: Vec<(u32, usize)> = Vec::new();
942            n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
943        }
944        n
945    }
946}
947impl Decode<'_> for SnapshotId {
948    fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
949        let flex = version >= 12;
950        let mut out = Self::default();
951        if version >= 0 {
952            out.end_offset = get_i64(buf)?;
953        }
954        if version >= 0 {
955            out.epoch = get_i32(buf)?;
956        }
957        if flex {
958            out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
959        }
960        Ok(out)
961    }
962}
963#[cfg(test)]
964impl SnapshotId {
965    #[must_use]
966    pub fn populated(version: i16) -> Self {
967        let mut m = Self::default();
968        if version >= 0 {
969            m.end_offset = 1i64;
970        }
971        if version >= 0 {
972            m.epoch = 1i32;
973        }
974        m
975    }
976}
977#[derive(Debug, Clone, PartialEq, Eq, Default)]
978pub struct AbortedTransaction {
979    pub producer_id: i64,
980    pub first_offset: i64,
981    pub unknown_tagged_fields: UnknownTaggedFields,
982}
983impl Encode for AbortedTransaction {
984    fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
985        let flex = version >= 12;
986        if version >= 4 {
987            put_i64(buf, self.producer_id);
988        }
989        if version >= 4 {
990            put_i64(buf, self.first_offset);
991        }
992        if flex {
993            let tagged = WriteTaggedFields::new();
994            tagged.write(buf, &self.unknown_tagged_fields)?;
995        }
996        Ok(())
997    }
998    fn encoded_len(&self, version: i16) -> usize {
999        let flex = version >= 12;
1000        let mut n: usize = 0;
1001        if version >= 4 {
1002            n += 8;
1003        }
1004        if version >= 4 {
1005            n += 8;
1006        }
1007        if flex {
1008            let known_pairs: Vec<(u32, usize)> = Vec::new();
1009            n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
1010        }
1011        n
1012    }
1013}
1014impl Decode<'_> for AbortedTransaction {
1015    fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
1016        let flex = version >= 12;
1017        let mut out = Self::default();
1018        if version >= 4 {
1019            out.producer_id = get_i64(buf)?;
1020        }
1021        if version >= 4 {
1022            out.first_offset = get_i64(buf)?;
1023        }
1024        if flex {
1025            out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
1026        }
1027        Ok(out)
1028    }
1029}
1030#[cfg(test)]
1031impl AbortedTransaction {
1032    #[must_use]
1033    pub fn populated(version: i16) -> Self {
1034        let mut m = Self::default();
1035        if version >= 4 {
1036            m.producer_id = 1i64;
1037        }
1038        if version >= 4 {
1039            m.first_offset = 1i64;
1040        }
1041        m
1042    }
1043}
1044#[derive(Debug, Clone, PartialEq, Eq, Default)]
1045pub struct NodeEndpoint {
1046    pub node_id: i32,
1047    pub host: String,
1048    pub port: i32,
1049    pub rack: Option<String>,
1050    pub unknown_tagged_fields: UnknownTaggedFields,
1051}
1052impl Encode for NodeEndpoint {
1053    fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
1054        let flex = version >= 12;
1055        if version >= 16 {
1056            put_i32(buf, self.node_id);
1057        }
1058        if version >= 16 {
1059            if flex {
1060                put_compact_string(buf, &self.host)?;
1061            } else {
1062                put_string(buf, &self.host)?;
1063            }
1064        }
1065        if version >= 16 {
1066            put_i32(buf, self.port);
1067        }
1068        if version >= 16 {
1069            if flex {
1070                put_compact_nullable_string(buf, self.rack.as_deref())?;
1071            } else {
1072                put_nullable_string(buf, self.rack.as_deref())?;
1073            }
1074        }
1075        if flex {
1076            let tagged = WriteTaggedFields::new();
1077            tagged.write(buf, &self.unknown_tagged_fields)?;
1078        }
1079        Ok(())
1080    }
1081    fn encoded_len(&self, version: i16) -> usize {
1082        let flex = version >= 12;
1083        let mut n: usize = 0;
1084        if version >= 16 {
1085            n += 4;
1086        }
1087        if version >= 16 {
1088            n += if flex {
1089                compact_string_len(&self.host)
1090            } else {
1091                string_len(&self.host)
1092            };
1093        }
1094        if version >= 16 {
1095            n += 4;
1096        }
1097        if version >= 16 {
1098            n += if flex {
1099                compact_nullable_string_len(self.rack.as_deref())
1100            } else {
1101                nullable_string_len(self.rack.as_deref())
1102            };
1103        }
1104        if flex {
1105            let known_pairs: Vec<(u32, usize)> = Vec::new();
1106            n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
1107        }
1108        n
1109    }
1110}
1111impl Decode<'_> for NodeEndpoint {
1112    fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
1113        let flex = version >= 12;
1114        let mut out = Self::default();
1115        if version >= 16 {
1116            out.node_id = get_i32(buf)?;
1117        }
1118        if version >= 16 {
1119            out.host = if flex {
1120                get_compact_string_owned(buf)?
1121            } else {
1122                get_string_owned(buf)?
1123            };
1124        }
1125        if version >= 16 {
1126            out.port = get_i32(buf)?;
1127        }
1128        if version >= 16 {
1129            out.rack = if flex {
1130                get_compact_nullable_string_owned(buf)?
1131            } else {
1132                get_nullable_string_owned(buf)?
1133            };
1134        }
1135        if flex {
1136            out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
1137        }
1138        Ok(out)
1139    }
1140}
1141#[cfg(test)]
1142impl NodeEndpoint {
1143    #[must_use]
1144    pub fn populated(version: i16) -> Self {
1145        let mut m = Self::default();
1146        if version >= 16 {
1147            m.node_id = 1i32;
1148        }
1149        if version >= 16 {
1150            m.host = "x".to_owned();
1151        }
1152        if version >= 16 {
1153            m.port = 1i32;
1154        }
1155        if version >= 16 {
1156            m.rack = Some("x".to_owned());
1157        }
1158        m
1159    }
1160}
1161/// Default JSON payload matching `Self::default()` for JVM oracle differential testing.
1162/// Only includes fields valid for the given version.
1163#[must_use]
1164#[allow(unused_comparisons)]
1165pub fn default_json(version: i16) -> ::serde_json::Value {
1166    let mut obj = ::serde_json::Map::new();
1167    if version >= 1 {
1168        obj.insert("throttleTimeMs".to_owned(), ::serde_json::json!(0));
1169    }
1170    if version >= 7 {
1171        obj.insert("errorCode".to_owned(), ::serde_json::json!(0));
1172    }
1173    if version >= 7 {
1174        obj.insert("sessionId".to_owned(), ::serde_json::json!(0));
1175    }
1176    obj.insert("responses".to_owned(), ::serde_json::Value::Array(vec![]));
1177    if version >= 16 {
1178        obj.insert(
1179            "nodeEndpoints".to_owned(),
1180            ::serde_json::Value::Array(vec![]),
1181        );
1182    }
1183    ::serde_json::Value::Object(obj)
1184}
1185/// A message with a non-default value in every tagged field, at every depth, for
1186/// the JVM oracle differential sweep. The value is the same at every version.
1187#[must_use]
1188pub fn tagged_fixture() -> FetchResponse {
1189    FetchResponse {
1190        responses: vec![FetchableTopicResponse {
1191            partitions: vec![PartitionData {
1192                diverging_epoch: EpochEndOffset {
1193                    epoch: 1i32,
1194                    end_offset: 1i64,
1195                    ..EpochEndOffset::default()
1196                },
1197                current_leader: LeaderIdAndEpoch {
1198                    leader_id: 1i32,
1199                    leader_epoch: 1i32,
1200                    ..LeaderIdAndEpoch::default()
1201                },
1202                snapshot_id: SnapshotId {
1203                    end_offset: 1i64,
1204                    epoch: 1i32,
1205                    ..SnapshotId::default()
1206                },
1207                ..PartitionData::default()
1208            }],
1209            ..FetchableTopicResponse::default()
1210        }],
1211        node_endpoints: vec![NodeEndpoint {
1212            node_id: 1i32,
1213            host: "x".to_owned(),
1214            port: 1i32,
1215            rack: Some("x".to_owned()),
1216            ..NodeEndpoint::default()
1217        }],
1218        ..FetchResponse::default()
1219    }
1220}
1221/// The JSON form of `tagged_fixture()` that Kafka's JSON converter reads at
1222/// `version`. It holds only the fields that the schema declares at that version.
1223#[must_use]
1224pub fn tagged_fixture_json(version: i16) -> ::serde_json::Value {
1225    let mut m = ::serde_json::Map::new();
1226    m.insert("throttleTimeMs".to_owned(), ::serde_json::json!(0));
1227    if version >= 7 {
1228        m.insert("errorCode".to_owned(), ::serde_json::json!(0));
1229    }
1230    if version >= 7 {
1231        m.insert("sessionId".to_owned(), ::serde_json::json!(0));
1232    }
1233    m.insert(
1234        "responses".to_owned(),
1235        ::serde_json::Value::Array(vec![{
1236            let mut m = ::serde_json::Map::new();
1237            if version <= 12 {
1238                m.insert(
1239                    "topic".to_owned(),
1240                    ::serde_json::Value::String(String::new()),
1241                );
1242            }
1243            if version >= 13 {
1244                m.insert(
1245                    "topicId".to_owned(),
1246                    ::serde_json::Value::String("AAAAAAAAAAAAAAAAAAAAAA".to_owned()),
1247                );
1248            }
1249            m.insert(
1250                "partitions".to_owned(),
1251                ::serde_json::Value::Array(vec![{
1252                    let mut m = ::serde_json::Map::new();
1253                    m.insert("partitionIndex".to_owned(), ::serde_json::json!(0));
1254                    m.insert("errorCode".to_owned(), ::serde_json::json!(0));
1255                    m.insert("highWatermark".to_owned(), ::serde_json::json!(0));
1256                    m.insert("lastStableOffset".to_owned(), ::serde_json::json!(-1));
1257                    if version >= 5 {
1258                        m.insert("logStartOffset".to_owned(), ::serde_json::json!(-1));
1259                    }
1260                    if version >= 12 {
1261                        m.insert("divergingEpoch".to_owned(), {
1262                            let mut m = ::serde_json::Map::new();
1263                            m.insert("epoch".to_owned(), ::serde_json::json!(1));
1264                            m.insert("endOffset".to_owned(), ::serde_json::json!(1));
1265                            ::serde_json::Value::Object(m)
1266                        });
1267                    }
1268                    if version >= 12 {
1269                        m.insert("currentLeader".to_owned(), {
1270                            let mut m = ::serde_json::Map::new();
1271                            m.insert("leaderId".to_owned(), ::serde_json::json!(1));
1272                            m.insert("leaderEpoch".to_owned(), ::serde_json::json!(1));
1273                            ::serde_json::Value::Object(m)
1274                        });
1275                    }
1276                    if version >= 12 {
1277                        m.insert("snapshotId".to_owned(), {
1278                            let mut m = ::serde_json::Map::new();
1279                            m.insert("endOffset".to_owned(), ::serde_json::json!(1));
1280                            m.insert("epoch".to_owned(), ::serde_json::json!(1));
1281                            ::serde_json::Value::Object(m)
1282                        });
1283                    }
1284                    m.insert(
1285                        "abortedTransactions".to_owned(),
1286                        ::serde_json::Value::Array(vec![]),
1287                    );
1288                    if version >= 11 {
1289                        m.insert("preferredReadReplica".to_owned(), ::serde_json::json!(-1));
1290                    }
1291                    m.insert("records".to_owned(), ::serde_json::Value::Null);
1292                    ::serde_json::Value::Object(m)
1293                }]),
1294            );
1295            ::serde_json::Value::Object(m)
1296        }]),
1297    );
1298    if version >= 16 {
1299        m.insert(
1300            "nodeEndpoints".to_owned(),
1301            ::serde_json::Value::Array(vec![{
1302                let mut m = ::serde_json::Map::new();
1303                m.insert("nodeId".to_owned(), ::serde_json::json!(1));
1304                m.insert(
1305                    "host".to_owned(),
1306                    ::serde_json::Value::String("x".to_owned()),
1307                );
1308                m.insert("port".to_owned(), ::serde_json::json!(1));
1309                m.insert(
1310                    "rack".to_owned(),
1311                    ::serde_json::Value::String("x".to_owned()),
1312                );
1313                ::serde_json::Value::Object(m)
1314            }]),
1315        );
1316    }
1317    ::serde_json::Value::Object(m)
1318}
krabka_protocol/opt/rustwide/workdir/generated/FetchResponse.owned.rs

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