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