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

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