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