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