1use crate::primitives::fixed::{get_i8, get_i32, get_i64, put_i8, put_i32, put_i64};
3use crate::primitives::string_bytes::{
4 compact_nullable_string_len, compact_string_len, nullable_string_len,
5 put_compact_nullable_string, put_compact_string, put_nullable_string, put_string, string_len,
6};
7use crate::primitives::string_bytes_borrowed::{get_compact_string_borrowed, get_string_borrowed};
8use crate::tagged_fields::{
9 WriteTaggedFields, encode_to_bytes, read_tagged_fields, tagged_fields_len,
10};
11use crate::{DecodeBorrow, Encode, ProtocolError, UnknownTaggedFields};
12use bytes::BufMut;
13pub const API_KEY: i16 = 1;
14pub const MIN_VERSION: i16 = 4;
15pub const MAX_VERSION: i16 = 18;
16pub const FLEXIBLE_MIN: i16 = 12;
17#[inline]
18fn is_flexible(version: i16) -> bool {
19 version >= FLEXIBLE_MIN
20}
21#[derive(Debug, Clone, PartialEq, Eq)]
22pub struct FetchRequest<'a> {
23 pub replica_id: i32,
24 pub max_wait_ms: i32,
25 pub min_bytes: i32,
26 pub max_bytes: i32,
27 pub isolation_level: i8,
28 pub session_id: i32,
29 pub session_epoch: i32,
30 pub topics: Vec<FetchTopic<'a>>,
31 pub forgotten_topics_data: Vec<ForgottenTopic<'a>>,
32 pub rack_id: &'a str,
33 pub cluster_id: Option<String>,
34 pub replica_state: ReplicaState,
35 pub unknown_tagged_fields: UnknownTaggedFields,
36}
37impl Default for FetchRequest<'_> {
38 fn default() -> Self {
39 Self {
40 replica_id: -1i32,
41 max_wait_ms: 0i32,
42 min_bytes: 0i32,
43 max_bytes: 2_147_483_647i32,
44 isolation_level: 0i8,
45 session_id: 0i32,
46 session_epoch: -1i32,
47 topics: Vec::new(),
48 forgotten_topics_data: Vec::new(),
49 rack_id: "",
50 cluster_id: None,
51 replica_state: Default::default(),
52 unknown_tagged_fields: Default::default(),
53 }
54 }
55}
56impl FetchRequest<'_> {
57 pub fn to_owned(&self) -> crate::owned::fetch_request::FetchRequest {
58 crate::owned::fetch_request::FetchRequest {
59 replica_id: (self.replica_id),
60 max_wait_ms: (self.max_wait_ms),
61 min_bytes: (self.min_bytes),
62 max_bytes: (self.max_bytes),
63 isolation_level: (self.isolation_level),
64 session_id: (self.session_id),
65 session_epoch: (self.session_epoch),
66 topics: (self.topics).iter().map(FetchTopic::to_owned).collect(),
67 forgotten_topics_data: (self.forgotten_topics_data)
68 .iter()
69 .map(ForgottenTopic::to_owned)
70 .collect(),
71 rack_id: (self.rack_id).to_string(),
72 cluster_id: self.cluster_id.clone(),
73 replica_state: (self.replica_state).to_owned(),
74 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
75 }
76 }
77}
78impl Encode for FetchRequest<'_> {
79 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
80 if !(MIN_VERSION..=MAX_VERSION).contains(&version) {
81 return Err(ProtocolError::UnsupportedVersion {
82 api_key: API_KEY,
83 version,
84 });
85 }
86 let flex = is_flexible(version);
87 if (0..=14).contains(&version) {
88 put_i32(buf, self.replica_id);
89 }
90 if version >= 0 {
91 put_i32(buf, self.max_wait_ms);
92 }
93 if version >= 0 {
94 put_i32(buf, self.min_bytes);
95 }
96 if version >= 3 {
97 put_i32(buf, self.max_bytes);
98 }
99 if version >= 4 {
100 put_i8(buf, self.isolation_level);
101 }
102 if version >= 7 {
103 put_i32(buf, self.session_id);
104 }
105 if version >= 7 {
106 put_i32(buf, self.session_epoch);
107 }
108 if version >= 0 {
109 {
110 crate::primitives::array::put_array_len(buf, (self.topics).len(), flex);
111 for it in &self.topics {
112 it.encode(buf, version)?;
113 }
114 }
115 }
116 if version >= 7 {
117 {
118 crate::primitives::array::put_array_len(
119 buf,
120 (self.forgotten_topics_data).len(),
121 flex,
122 );
123 for it in &self.forgotten_topics_data {
124 it.encode(buf, version)?;
125 }
126 }
127 }
128 if version >= 11 {
129 if flex {
130 put_compact_string(buf, self.rack_id);
131 } else {
132 put_string(buf, self.rack_id);
133 }
134 }
135 if flex {
136 let mut tagged = WriteTaggedFields::new();
137 if !(self.cluster_id.is_none()) {
138 let payload = encode_to_bytes(
139 if flex {
140 compact_nullable_string_len(self.cluster_id.as_deref())
141 } else {
142 nullable_string_len(self.cluster_id.as_deref())
143 },
144 |b| {
145 if flex {
146 put_compact_nullable_string(b, self.cluster_id.as_deref());
147 } else {
148 put_nullable_string(b, self.cluster_id.as_deref());
149 }
150 Ok(())
151 },
152 );
153 tagged.add(0, payload);
154 }
155 if !(crate::codegen_helpers::is_default(&self.replica_state)) {
156 let payload = encode_to_bytes(self.replica_state.encoded_len(version), |b| {
157 self.replica_state.encode(b, version)?;
158 Ok(())
159 });
160 tagged.add(1, payload);
161 }
162 tagged.write(buf, &self.unknown_tagged_fields);
163 }
164 Ok(())
165 }
166 fn encoded_len(&self, version: i16) -> usize {
167 let flex = is_flexible(version);
168 let mut n: usize = 0;
169 if (0..=14).contains(&version) {
170 n += 4;
171 }
172 if version >= 0 {
173 n += 4;
174 }
175 if version >= 0 {
176 n += 4;
177 }
178 if version >= 3 {
179 n += 4;
180 }
181 if version >= 4 {
182 n += 1;
183 }
184 if version >= 7 {
185 n += 4;
186 }
187 if version >= 7 {
188 n += 4;
189 }
190 if version >= 0 {
191 n += {
192 let prefix =
193 crate::primitives::array::array_len_prefix_len((self.topics).len(), flex);
194 let body: usize = (self.topics).iter().map(|it| it.encoded_len(version)).sum();
195 prefix + body
196 };
197 }
198 if version >= 7 {
199 n += {
200 let prefix = crate::primitives::array::array_len_prefix_len(
201 (self.forgotten_topics_data).len(),
202 flex,
203 );
204 let body: usize = (self.forgotten_topics_data)
205 .iter()
206 .map(|it| it.encoded_len(version))
207 .sum();
208 prefix + body
209 };
210 }
211 if version >= 11 {
212 n += if flex {
213 compact_string_len(self.rack_id)
214 } else {
215 string_len(self.rack_id)
216 };
217 }
218 if flex {
219 let mut known_pairs: Vec<(u32, usize)> = Vec::new();
220 if !(self.cluster_id.is_none()) {
221 known_pairs.push((
222 0,
223 if flex {
224 compact_nullable_string_len(self.cluster_id.as_deref())
225 } else {
226 nullable_string_len(self.cluster_id.as_deref())
227 },
228 ));
229 }
230 if !(crate::codegen_helpers::is_default(&self.replica_state)) {
231 known_pairs.push((1, self.replica_state.encoded_len(version)));
232 }
233 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
234 }
235 n
236 }
237}
238impl<'de> DecodeBorrow<'de> for FetchRequest<'de> {
239 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
240 if !(MIN_VERSION..=MAX_VERSION).contains(&version) {
241 return Err(ProtocolError::UnsupportedVersion {
242 api_key: API_KEY,
243 version,
244 });
245 }
246 let flex = is_flexible(version);
247 let mut out = Self::default();
248 if (0..=14).contains(&version) {
249 out.replica_id = get_i32(buf)?;
250 }
251 if version >= 0 {
252 out.max_wait_ms = get_i32(buf)?;
253 }
254 if version >= 0 {
255 out.min_bytes = get_i32(buf)?;
256 }
257 if version >= 3 {
258 out.max_bytes = get_i32(buf)?;
259 }
260 if version >= 4 {
261 out.isolation_level = get_i8(buf)?;
262 }
263 if version >= 7 {
264 out.session_id = get_i32(buf)?;
265 }
266 if version >= 7 {
267 out.session_epoch = get_i32(buf)?;
268 }
269 if version >= 0 {
270 out.topics = {
271 let n = crate::primitives::array::get_array_len(buf, flex)?;
272 let mut v = Vec::with_capacity(n);
273 for _ in 0..n {
274 v.push(FetchTopic::decode_borrow(buf, version)?);
275 }
276 v
277 };
278 }
279 if version >= 7 {
280 out.forgotten_topics_data = {
281 let n = crate::primitives::array::get_array_len(buf, flex)?;
282 let mut v = Vec::with_capacity(n);
283 for _ in 0..n {
284 v.push(ForgottenTopic::decode_borrow(buf, version)?);
285 }
286 v
287 };
288 }
289 if version >= 11 {
290 out.rack_id = if flex {
291 get_compact_string_borrowed(buf)?
292 } else {
293 get_string_borrowed(buf)?
294 };
295 }
296 if flex {
297 let mut tag_cluster_id = None;
298 let mut tag_replica_state = None;
299 out.unknown_tagged_fields = read_tagged_fields(buf, |tag, payload| match tag {
300 0 => {
301 tag_cluster_id = Some({
302 let b: &mut &[u8] = payload;
303 if flex {
304 crate::primitives::string_bytes::get_compact_nullable_string_owned(b)?
305 } else {
306 crate::primitives::string_bytes::get_nullable_string_owned(b)?
307 }
308 });
309 Ok(true)
310 }
311 1 => {
312 tag_replica_state = Some({
313 let b: &mut &[u8] = payload;
314 ReplicaState::decode_borrow(b, version)?
315 });
316 Ok(true)
317 }
318 _ => Ok(false),
319 })?;
320 if let Some(v) = tag_cluster_id {
321 out.cluster_id = v;
322 }
323 if let Some(v) = tag_replica_state {
324 out.replica_state = v;
325 }
326 }
327 Ok(out)
328 }
329}
330#[cfg(test)]
331impl FetchRequest<'_> {
332 #[must_use]
333 pub fn populated(version: i16) -> Self {
334 let mut m = Self::default();
335 if (0..=14).contains(&version) {
336 m.replica_id = 1i32;
337 }
338 if version >= 0 {
339 m.max_wait_ms = 1i32;
340 }
341 if version >= 0 {
342 m.min_bytes = 1i32;
343 }
344 if version >= 3 {
345 m.max_bytes = 1i32;
346 }
347 if version >= 4 {
348 m.isolation_level = 1i8;
349 }
350 if version >= 7 {
351 m.session_id = 1i32;
352 }
353 if version >= 7 {
354 m.session_epoch = 1i32;
355 }
356 if version >= 0 {
357 m.topics = vec![FetchTopic::populated(version)];
358 }
359 if version >= 7 {
360 m.forgotten_topics_data = vec![ForgottenTopic::populated(version)];
361 }
362 if version >= 11 {
363 m.rack_id = "x";
364 }
365 if version >= 12 {
366 m.cluster_id = Some("x".to_string());
367 }
368 if version >= 15 {
369 m.replica_state = ReplicaState::populated(version);
370 }
371 m
372 }
373}
374#[derive(Debug, Clone, PartialEq, Eq)]
375pub struct ReplicaState {
376 pub replica_id: i32,
377 pub replica_epoch: i64,
378 pub unknown_tagged_fields: UnknownTaggedFields,
379}
380impl Default for ReplicaState {
381 fn default() -> Self {
382 Self {
383 replica_id: -1i32,
384 replica_epoch: -1i64,
385 unknown_tagged_fields: Default::default(),
386 }
387 }
388}
389impl ReplicaState {
390 pub fn to_owned(&self) -> crate::owned::fetch_request::ReplicaState {
391 crate::owned::fetch_request::ReplicaState {
392 replica_id: (self.replica_id),
393 replica_epoch: (self.replica_epoch),
394 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
395 }
396 }
397}
398impl Encode for ReplicaState {
399 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
400 let flex = version >= 12;
401 if version >= 15 {
402 put_i32(buf, self.replica_id);
403 }
404 if version >= 15 {
405 put_i64(buf, self.replica_epoch);
406 }
407 if flex {
408 let tagged = WriteTaggedFields::new();
409 tagged.write(buf, &self.unknown_tagged_fields);
410 }
411 Ok(())
412 }
413 fn encoded_len(&self, version: i16) -> usize {
414 let flex = version >= 12;
415 let mut n: usize = 0;
416 if version >= 15 {
417 n += 4;
418 }
419 if version >= 15 {
420 n += 8;
421 }
422 if flex {
423 let known_pairs: Vec<(u32, usize)> = Vec::new();
424 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
425 }
426 n
427 }
428}
429impl<'de> DecodeBorrow<'de> for ReplicaState {
430 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
431 let flex = version >= 12;
432 let mut out = Self::default();
433 if version >= 15 {
434 out.replica_id = get_i32(buf)?;
435 }
436 if version >= 15 {
437 out.replica_epoch = get_i64(buf)?;
438 }
439 if flex {
440 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
441 }
442 Ok(out)
443 }
444}
445#[cfg(test)]
446impl ReplicaState {
447 #[must_use]
448 pub fn populated(version: i16) -> Self {
449 let mut m = Self::default();
450 if version >= 15 {
451 m.replica_id = 1i32;
452 }
453 if version >= 15 {
454 m.replica_epoch = 1i64;
455 }
456 m
457 }
458}
459#[derive(Debug, Clone, PartialEq, Eq, Default)]
460pub struct FetchTopic<'a> {
461 pub topic: &'a str,
462 pub topic_id: crate::primitives::uuid::Uuid,
463 pub partitions: Vec<FetchPartition>,
464 pub unknown_tagged_fields: UnknownTaggedFields,
465}
466impl FetchTopic<'_> {
467 pub fn to_owned(&self) -> crate::owned::fetch_request::FetchTopic {
468 crate::owned::fetch_request::FetchTopic {
469 topic: (self.topic).to_string(),
470 topic_id: (self.topic_id),
471 partitions: (self.partitions)
472 .iter()
473 .map(FetchPartition::to_owned)
474 .collect(),
475 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
476 }
477 }
478}
479impl Encode for FetchTopic<'_> {
480 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
481 let flex = version >= 12;
482 if (0..=12).contains(&version) {
483 if flex {
484 put_compact_string(buf, self.topic);
485 } else {
486 put_string(buf, self.topic);
487 }
488 }
489 if version >= 13 {
490 crate::primitives::uuid::put_uuid(buf, self.topic_id);
491 }
492 if version >= 0 {
493 {
494 crate::primitives::array::put_array_len(buf, (self.partitions).len(), flex);
495 for it in &self.partitions {
496 it.encode(buf, version)?;
497 }
498 }
499 }
500 if flex {
501 let tagged = WriteTaggedFields::new();
502 tagged.write(buf, &self.unknown_tagged_fields);
503 }
504 Ok(())
505 }
506 fn encoded_len(&self, version: i16) -> usize {
507 let flex = version >= 12;
508 let mut n: usize = 0;
509 if (0..=12).contains(&version) {
510 n += if flex {
511 compact_string_len(self.topic)
512 } else {
513 string_len(self.topic)
514 };
515 }
516 if version >= 13 {
517 n += 16;
518 }
519 if version >= 0 {
520 n += {
521 let prefix =
522 crate::primitives::array::array_len_prefix_len((self.partitions).len(), flex);
523 let body: usize = (self.partitions)
524 .iter()
525 .map(|it| it.encoded_len(version))
526 .sum();
527 prefix + body
528 };
529 }
530 if flex {
531 let known_pairs: Vec<(u32, usize)> = Vec::new();
532 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
533 }
534 n
535 }
536}
537impl<'de> DecodeBorrow<'de> for FetchTopic<'de> {
538 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
539 let flex = version >= 12;
540 let mut out = Self::default();
541 if (0..=12).contains(&version) {
542 out.topic = if flex {
543 get_compact_string_borrowed(buf)?
544 } else {
545 get_string_borrowed(buf)?
546 };
547 }
548 if version >= 13 {
549 out.topic_id = crate::primitives::uuid::get_uuid(buf)?;
550 }
551 if version >= 0 {
552 out.partitions = {
553 let n = crate::primitives::array::get_array_len(buf, flex)?;
554 let mut v = Vec::with_capacity(n);
555 for _ in 0..n {
556 v.push(FetchPartition::decode_borrow(buf, version)?);
557 }
558 v
559 };
560 }
561 if flex {
562 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
563 }
564 Ok(out)
565 }
566}
567#[cfg(test)]
568impl FetchTopic<'_> {
569 #[must_use]
570 pub fn populated(version: i16) -> Self {
571 let mut m = Self::default();
572 if (0..=12).contains(&version) {
573 m.topic = "x";
574 }
575 if version >= 13 {
576 m.topic_id = crate::primitives::uuid::Uuid([1u8; 16]);
577 }
578 if version >= 0 {
579 m.partitions = vec![FetchPartition::populated(version)];
580 }
581 m
582 }
583}
584#[derive(Debug, Clone, PartialEq, Eq)]
585pub struct FetchPartition {
586 pub partition: i32,
587 pub current_leader_epoch: i32,
588 pub fetch_offset: i64,
589 pub last_fetched_epoch: i32,
590 pub log_start_offset: i64,
591 pub partition_max_bytes: i32,
592 pub replica_directory_id: crate::primitives::uuid::Uuid,
593 pub high_watermark: i64,
594 pub unknown_tagged_fields: UnknownTaggedFields,
595}
596impl Default for FetchPartition {
597 fn default() -> Self {
598 Self {
599 partition: 0i32,
600 current_leader_epoch: -1i32,
601 fetch_offset: 0i64,
602 last_fetched_epoch: -1i32,
603 log_start_offset: -1i64,
604 partition_max_bytes: 0i32,
605 replica_directory_id: Default::default(),
606 high_watermark: 9_223_372_036_854_775_807i64,
607 unknown_tagged_fields: Default::default(),
608 }
609 }
610}
611impl FetchPartition {
612 pub fn to_owned(&self) -> crate::owned::fetch_request::FetchPartition {
613 crate::owned::fetch_request::FetchPartition {
614 partition: (self.partition),
615 current_leader_epoch: (self.current_leader_epoch),
616 fetch_offset: (self.fetch_offset),
617 last_fetched_epoch: (self.last_fetched_epoch),
618 log_start_offset: (self.log_start_offset),
619 partition_max_bytes: (self.partition_max_bytes),
620 replica_directory_id: (self.replica_directory_id),
621 high_watermark: (self.high_watermark),
622 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
623 }
624 }
625}
626impl Encode for FetchPartition {
627 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
628 let flex = version >= 12;
629 if version >= 0 {
630 put_i32(buf, self.partition);
631 }
632 if version >= 9 {
633 put_i32(buf, self.current_leader_epoch);
634 }
635 if version >= 0 {
636 put_i64(buf, self.fetch_offset);
637 }
638 if version >= 12 {
639 put_i32(buf, self.last_fetched_epoch);
640 }
641 if version >= 5 {
642 put_i64(buf, self.log_start_offset);
643 }
644 if version >= 0 {
645 put_i32(buf, self.partition_max_bytes);
646 }
647 if flex {
648 let mut tagged = WriteTaggedFields::new();
649 if !(crate::codegen_helpers::is_default(&self.replica_directory_id)) {
650 let payload = encode_to_bytes(16, |b| {
651 crate::primitives::uuid::put_uuid(b, self.replica_directory_id);
652 Ok(())
653 });
654 tagged.add(0, payload);
655 }
656 if !(self.high_watermark == 9_223_372_036_854_775_807i64) {
657 let payload = encode_to_bytes(8, |b| {
658 put_i64(b, self.high_watermark);
659 Ok(())
660 });
661 tagged.add(1, payload);
662 }
663 tagged.write(buf, &self.unknown_tagged_fields);
664 }
665 Ok(())
666 }
667 fn encoded_len(&self, version: i16) -> usize {
668 let flex = version >= 12;
669 let mut n: usize = 0;
670 if version >= 0 {
671 n += 4;
672 }
673 if version >= 9 {
674 n += 4;
675 }
676 if version >= 0 {
677 n += 8;
678 }
679 if version >= 12 {
680 n += 4;
681 }
682 if version >= 5 {
683 n += 8;
684 }
685 if version >= 0 {
686 n += 4;
687 }
688 if flex {
689 let mut known_pairs: Vec<(u32, usize)> = Vec::new();
690 if !(crate::codegen_helpers::is_default(&self.replica_directory_id)) {
691 known_pairs.push((0, 16));
692 }
693 if !(self.high_watermark == 9_223_372_036_854_775_807i64) {
694 known_pairs.push((1, 8));
695 }
696 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
697 }
698 n
699 }
700}
701impl<'de> DecodeBorrow<'de> for FetchPartition {
702 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
703 let flex = version >= 12;
704 let mut out = Self::default();
705 if version >= 0 {
706 out.partition = get_i32(buf)?;
707 }
708 if version >= 9 {
709 out.current_leader_epoch = get_i32(buf)?;
710 }
711 if version >= 0 {
712 out.fetch_offset = get_i64(buf)?;
713 }
714 if version >= 12 {
715 out.last_fetched_epoch = get_i32(buf)?;
716 }
717 if version >= 5 {
718 out.log_start_offset = get_i64(buf)?;
719 }
720 if version >= 0 {
721 out.partition_max_bytes = get_i32(buf)?;
722 }
723 if flex {
724 let mut tag_replica_directory_id = None;
725 let mut tag_high_watermark = None;
726 out.unknown_tagged_fields = read_tagged_fields(buf, |tag, payload| match tag {
727 0 => {
728 tag_replica_directory_id = Some({
729 let b: &mut &[u8] = payload;
730 crate::primitives::uuid::get_uuid(b)?
731 });
732 Ok(true)
733 }
734 1 => {
735 tag_high_watermark = Some({
736 let b: &mut &[u8] = payload;
737 get_i64(b)?
738 });
739 Ok(true)
740 }
741 _ => Ok(false),
742 })?;
743 if let Some(v) = tag_replica_directory_id {
744 out.replica_directory_id = v;
745 }
746 if let Some(v) = tag_high_watermark {
747 out.high_watermark = v;
748 }
749 }
750 Ok(out)
751 }
752}
753#[cfg(test)]
754impl FetchPartition {
755 #[must_use]
756 pub fn populated(version: i16) -> Self {
757 let mut m = Self::default();
758 if version >= 0 {
759 m.partition = 1i32;
760 }
761 if version >= 9 {
762 m.current_leader_epoch = 1i32;
763 }
764 if version >= 0 {
765 m.fetch_offset = 1i64;
766 }
767 if version >= 12 {
768 m.last_fetched_epoch = 1i32;
769 }
770 if version >= 5 {
771 m.log_start_offset = 1i64;
772 }
773 if version >= 0 {
774 m.partition_max_bytes = 1i32;
775 }
776 if version >= 17 {
777 m.replica_directory_id = crate::primitives::uuid::Uuid([1u8; 16]);
778 }
779 if version >= 18 {
780 m.high_watermark = 1i64;
781 }
782 m
783 }
784}
785#[derive(Debug, Clone, PartialEq, Eq, Default)]
786pub struct ForgottenTopic<'a> {
787 pub topic: &'a str,
788 pub topic_id: crate::primitives::uuid::Uuid,
789 pub partitions: Vec<i32>,
790 pub unknown_tagged_fields: UnknownTaggedFields,
791}
792impl ForgottenTopic<'_> {
793 pub fn to_owned(&self) -> crate::owned::fetch_request::ForgottenTopic {
794 crate::owned::fetch_request::ForgottenTopic {
795 topic: (self.topic).to_string(),
796 topic_id: (self.topic_id),
797 partitions: (self.partitions).clone(),
798 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
799 }
800 }
801}
802impl Encode for ForgottenTopic<'_> {
803 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
804 let flex = version >= 12;
805 if (7..=12).contains(&version) {
806 if flex {
807 put_compact_string(buf, self.topic);
808 } else {
809 put_string(buf, self.topic);
810 }
811 }
812 if version >= 13 {
813 crate::primitives::uuid::put_uuid(buf, self.topic_id);
814 }
815 if version >= 7 {
816 {
817 crate::primitives::array::put_array_len(buf, (self.partitions).len(), flex);
818 for it in &self.partitions {
819 put_i32(buf, *it);
820 }
821 }
822 }
823 if flex {
824 let tagged = WriteTaggedFields::new();
825 tagged.write(buf, &self.unknown_tagged_fields);
826 }
827 Ok(())
828 }
829 fn encoded_len(&self, version: i16) -> usize {
830 let flex = version >= 12;
831 let mut n: usize = 0;
832 if (7..=12).contains(&version) {
833 n += if flex {
834 compact_string_len(self.topic)
835 } else {
836 string_len(self.topic)
837 };
838 }
839 if version >= 13 {
840 n += 16;
841 }
842 if version >= 7 {
843 n += {
844 let prefix =
845 crate::primitives::array::array_len_prefix_len((self.partitions).len(), flex);
846 let body: usize = (self.partitions).iter().map(|_| 4).sum();
847 prefix + body
848 };
849 }
850 if flex {
851 let known_pairs: Vec<(u32, usize)> = Vec::new();
852 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
853 }
854 n
855 }
856}
857impl<'de> DecodeBorrow<'de> for ForgottenTopic<'de> {
858 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
859 let flex = version >= 12;
860 let mut out = Self::default();
861 if (7..=12).contains(&version) {
862 out.topic = if flex {
863 get_compact_string_borrowed(buf)?
864 } else {
865 get_string_borrowed(buf)?
866 };
867 }
868 if version >= 13 {
869 out.topic_id = crate::primitives::uuid::get_uuid(buf)?;
870 }
871 if version >= 7 {
872 out.partitions = {
873 let n = crate::primitives::array::get_array_len(buf, flex)?;
874 let mut v = Vec::with_capacity(n);
875 for _ in 0..n {
876 v.push(get_i32(buf)?);
877 }
878 v
879 };
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 ForgottenTopic<'_> {
889 #[must_use]
890 pub fn populated(version: i16) -> Self {
891 let mut m = Self::default();
892 if (7..=12).contains(&version) {
893 m.topic = "x";
894 }
895 if version >= 13 {
896 m.topic_id = crate::primitives::uuid::Uuid([1u8; 16]);
897 }
898 if version >= 7 {
899 m.partitions = vec![1i32];
900 }
901 m
902 }
903}