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

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