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