1use bytes::BufMut;
4
5use crate::primitives::fixed::{get_i32, get_u16, put_i32, put_u16};
6use crate::primitives::string_bytes::{
7 compact_nullable_string_len, compact_string_len, nullable_string_len,
8 put_compact_nullable_string, put_compact_string, put_nullable_string, put_string, string_len,
9};
10use crate::primitives::string_bytes_borrowed::{
11 get_compact_nullable_string_borrowed, get_compact_string_borrowed,
12 get_nullable_string_borrowed, get_string_borrowed,
13};
14use crate::tagged_fields::{WriteTaggedFields, read_tagged_fields, tagged_fields_len};
15use crate::{DecodeBorrow, Encode, ProtocolError, UnknownTaggedFields};
16
17pub const API_KEY: i16 = 54;
18pub const MIN_VERSION: i16 = 0;
19pub const MAX_VERSION: i16 = 1;
20pub const FLEXIBLE_MIN: i16 = 1;
21
22#[inline]
23fn is_flexible(version: i16) -> bool {
24 version >= FLEXIBLE_MIN
25}
26
27#[derive(Debug, Clone, PartialEq, Eq, Default)]
28pub struct EndQuorumEpochRequest<'a> {
29 pub cluster_id: Option<&'a str>,
30 pub topics: Vec<TopicData<'a>>,
31 pub leader_endpoints: Vec<LeaderEndpoint<'a>>,
32 pub unknown_tagged_fields: UnknownTaggedFields,
33}
34impl EndQuorumEpochRequest<'_> {
35 pub fn to_owned(&self) -> crate::owned::end_quorum_epoch_request::EndQuorumEpochRequest {
36 crate::owned::end_quorum_epoch_request::EndQuorumEpochRequest {
37 cluster_id: (self.cluster_id).map(std::string::ToString::to_string),
38 topics: (self.topics).iter().map(TopicData::to_owned).collect(),
39 leader_endpoints: (self.leader_endpoints)
40 .iter()
41 .map(LeaderEndpoint::to_owned)
42 .collect(),
43 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
44 }
45 }
46}
47impl Encode for EndQuorumEpochRequest<'_> {
48 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
49 if !(MIN_VERSION..=MAX_VERSION).contains(&version) {
50 return Err(ProtocolError::UnsupportedVersion {
51 api_key: API_KEY,
52 version,
53 });
54 }
55 let flex = is_flexible(version);
56 if version >= 0 {
57 if flex {
58 put_compact_nullable_string(buf, self.cluster_id);
59 } else {
60 put_nullable_string(buf, self.cluster_id);
61 }
62 }
63 if version >= 0 {
64 {
65 crate::primitives::array::put_array_len(buf, (self.topics).len(), flex);
66 for it in &self.topics {
67 it.encode(buf, version)?;
68 }
69 }
70 }
71 if version >= 1 {
72 {
73 crate::primitives::array::put_array_len(buf, (self.leader_endpoints).len(), flex);
74 for it in &self.leader_endpoints {
75 it.encode(buf, version)?;
76 }
77 }
78 }
79 if flex {
80 let tagged = WriteTaggedFields::new();
81 tagged.write(buf, &self.unknown_tagged_fields);
82 }
83 Ok(())
84 }
85 fn encoded_len(&self, version: i16) -> usize {
86 let flex = is_flexible(version);
87 let mut n: usize = 0;
88 if version >= 0 {
89 n += if flex {
90 compact_nullable_string_len(self.cluster_id)
91 } else {
92 nullable_string_len(self.cluster_id)
93 };
94 }
95 if version >= 0 {
96 n += {
97 let prefix =
98 crate::primitives::array::array_len_prefix_len((self.topics).len(), flex);
99 let body: usize = (self.topics).iter().map(|it| it.encoded_len(version)).sum();
100 prefix + body
101 };
102 }
103 if version >= 1 {
104 n += {
105 let prefix = crate::primitives::array::array_len_prefix_len(
106 (self.leader_endpoints).len(),
107 flex,
108 );
109 let body: usize = (self.leader_endpoints)
110 .iter()
111 .map(|it| it.encoded_len(version))
112 .sum();
113 prefix + body
114 };
115 }
116 if flex {
117 let known_pairs: Vec<(u32, usize)> = Vec::new();
118 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
119 }
120 n
121 }
122}
123impl<'de> DecodeBorrow<'de> for EndQuorumEpochRequest<'de> {
124 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
125 if !(MIN_VERSION..=MAX_VERSION).contains(&version) {
126 return Err(ProtocolError::UnsupportedVersion {
127 api_key: API_KEY,
128 version,
129 });
130 }
131 let flex = is_flexible(version);
132 let mut out = Self::default();
133 if version >= 0 {
134 out.cluster_id = if flex {
135 get_compact_nullable_string_borrowed(buf)?
136 } else {
137 get_nullable_string_borrowed(buf)?
138 };
139 }
140 if version >= 0 {
141 out.topics = {
142 let n = crate::primitives::array::get_array_len(buf, flex)?;
143 let mut v = Vec::with_capacity(n);
144 for _ in 0..n {
145 v.push(TopicData::decode_borrow(buf, version)?);
146 }
147 v
148 };
149 }
150 if version >= 1 {
151 out.leader_endpoints = {
152 let n = crate::primitives::array::get_array_len(buf, flex)?;
153 let mut v = Vec::with_capacity(n);
154 for _ in 0..n {
155 v.push(LeaderEndpoint::decode_borrow(buf, version)?);
156 }
157 v
158 };
159 }
160 if flex {
161 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
162 }
163 Ok(out)
164 }
165}
166#[cfg(test)]
167impl EndQuorumEpochRequest<'_> {
168 #[must_use]
169 pub fn populated(version: i16) -> Self {
170 let mut m = Self::default();
171 if version >= 0 {
172 m.cluster_id = Some("x");
173 }
174 if version >= 0 {
175 m.topics = vec![TopicData::populated(version)];
176 }
177 if version >= 1 {
178 m.leader_endpoints = vec![LeaderEndpoint::populated(version)];
179 }
180 m
181 }
182}
183#[derive(Debug, Clone, PartialEq, Eq, Default)]
184pub struct TopicData<'a> {
185 pub topic_name: &'a str,
186 pub partitions: Vec<PartitionData>,
187 pub unknown_tagged_fields: UnknownTaggedFields,
188}
189impl TopicData<'_> {
190 pub fn to_owned(&self) -> crate::owned::end_quorum_epoch_request::TopicData {
191 crate::owned::end_quorum_epoch_request::TopicData {
192 topic_name: (self.topic_name).to_string(),
193 partitions: (self.partitions)
194 .iter()
195 .map(PartitionData::to_owned)
196 .collect(),
197 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
198 }
199 }
200}
201impl Encode for TopicData<'_> {
202 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
203 let flex = version >= 1;
204 if version >= 0 {
205 if flex {
206 put_compact_string(buf, self.topic_name);
207 } else {
208 put_string(buf, self.topic_name);
209 }
210 }
211 if version >= 0 {
212 {
213 crate::primitives::array::put_array_len(buf, (self.partitions).len(), flex);
214 for it in &self.partitions {
215 it.encode(buf, version)?;
216 }
217 }
218 }
219 if flex {
220 let tagged = WriteTaggedFields::new();
221 tagged.write(buf, &self.unknown_tagged_fields);
222 }
223 Ok(())
224 }
225 fn encoded_len(&self, version: i16) -> usize {
226 let flex = version >= 1;
227 let mut n: usize = 0;
228 if version >= 0 {
229 n += if flex {
230 compact_string_len(self.topic_name)
231 } else {
232 string_len(self.topic_name)
233 };
234 }
235 if version >= 0 {
236 n += {
237 let prefix =
238 crate::primitives::array::array_len_prefix_len((self.partitions).len(), flex);
239 let body: usize = (self.partitions)
240 .iter()
241 .map(|it| it.encoded_len(version))
242 .sum();
243 prefix + body
244 };
245 }
246 if flex {
247 let known_pairs: Vec<(u32, usize)> = Vec::new();
248 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
249 }
250 n
251 }
252}
253impl<'de> DecodeBorrow<'de> for TopicData<'de> {
254 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
255 let flex = version >= 1;
256 let mut out = Self::default();
257 if version >= 0 {
258 out.topic_name = if flex {
259 get_compact_string_borrowed(buf)?
260 } else {
261 get_string_borrowed(buf)?
262 };
263 }
264 if version >= 0 {
265 out.partitions = {
266 let n = crate::primitives::array::get_array_len(buf, flex)?;
267 let mut v = Vec::with_capacity(n);
268 for _ in 0..n {
269 v.push(PartitionData::decode_borrow(buf, version)?);
270 }
271 v
272 };
273 }
274 if flex {
275 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
276 }
277 Ok(out)
278 }
279}
280#[cfg(test)]
281impl TopicData<'_> {
282 #[must_use]
283 pub fn populated(version: i16) -> Self {
284 let mut m = Self::default();
285 if version >= 0 {
286 m.topic_name = "x";
287 }
288 if version >= 0 {
289 m.partitions = vec![PartitionData::populated(version)];
290 }
291 m
292 }
293}
294#[derive(Debug, Clone, PartialEq, Eq, Default)]
295pub struct PartitionData {
296 pub partition_index: i32,
297 pub leader_id: i32,
298 pub leader_epoch: i32,
299 pub preferred_successors: Vec<i32>,
300 pub preferred_candidates: Vec<ReplicaInfo>,
301 pub unknown_tagged_fields: UnknownTaggedFields,
302}
303impl PartitionData {
304 pub fn to_owned(&self) -> crate::owned::end_quorum_epoch_request::PartitionData {
305 crate::owned::end_quorum_epoch_request::PartitionData {
306 partition_index: (self.partition_index),
307 leader_id: (self.leader_id),
308 leader_epoch: (self.leader_epoch),
309 preferred_successors: (self.preferred_successors).clone(),
310 preferred_candidates: (self.preferred_candidates)
311 .iter()
312 .map(ReplicaInfo::to_owned)
313 .collect(),
314 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
315 }
316 }
317}
318impl Encode for PartitionData {
319 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
320 let flex = version >= 1;
321 if version >= 0 {
322 put_i32(buf, self.partition_index);
323 }
324 if version >= 0 {
325 put_i32(buf, self.leader_id);
326 }
327 if version >= 0 {
328 put_i32(buf, self.leader_epoch);
329 }
330 if version == 0 {
331 {
332 crate::primitives::array::put_array_len(
333 buf,
334 (self.preferred_successors).len(),
335 flex,
336 );
337 for it in &self.preferred_successors {
338 put_i32(buf, *it);
339 }
340 }
341 }
342 if version >= 1 {
343 {
344 crate::primitives::array::put_array_len(
345 buf,
346 (self.preferred_candidates).len(),
347 flex,
348 );
349 for it in &self.preferred_candidates {
350 it.encode(buf, version)?;
351 }
352 }
353 }
354 if flex {
355 let tagged = WriteTaggedFields::new();
356 tagged.write(buf, &self.unknown_tagged_fields);
357 }
358 Ok(())
359 }
360 fn encoded_len(&self, version: i16) -> usize {
361 let flex = version >= 1;
362 let mut n: usize = 0;
363 if version >= 0 {
364 n += 4;
365 }
366 if version >= 0 {
367 n += 4;
368 }
369 if version >= 0 {
370 n += 4;
371 }
372 if version == 0 {
373 n += {
374 let prefix = crate::primitives::array::array_len_prefix_len(
375 (self.preferred_successors).len(),
376 flex,
377 );
378 let body: usize = (self.preferred_successors).iter().map(|_| 4).sum();
379 prefix + body
380 };
381 }
382 if version >= 1 {
383 n += {
384 let prefix = crate::primitives::array::array_len_prefix_len(
385 (self.preferred_candidates).len(),
386 flex,
387 );
388 let body: usize = (self.preferred_candidates)
389 .iter()
390 .map(|it| it.encoded_len(version))
391 .sum();
392 prefix + body
393 };
394 }
395 if flex {
396 let known_pairs: Vec<(u32, usize)> = Vec::new();
397 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
398 }
399 n
400 }
401}
402impl<'de> DecodeBorrow<'de> for PartitionData {
403 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
404 let flex = version >= 1;
405 let mut out = Self::default();
406 if version >= 0 {
407 out.partition_index = get_i32(buf)?;
408 }
409 if version >= 0 {
410 out.leader_id = get_i32(buf)?;
411 }
412 if version >= 0 {
413 out.leader_epoch = get_i32(buf)?;
414 }
415 if version == 0 {
416 out.preferred_successors = {
417 let n = crate::primitives::array::get_array_len(buf, flex)?;
418 let mut v = Vec::with_capacity(n);
419 for _ in 0..n {
420 v.push(get_i32(buf)?);
421 }
422 v
423 };
424 }
425 if version >= 1 {
426 out.preferred_candidates = {
427 let n = crate::primitives::array::get_array_len(buf, flex)?;
428 let mut v = Vec::with_capacity(n);
429 for _ in 0..n {
430 v.push(ReplicaInfo::decode_borrow(buf, version)?);
431 }
432 v
433 };
434 }
435 if flex {
436 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
437 }
438 Ok(out)
439 }
440}
441#[cfg(test)]
442impl PartitionData {
443 #[must_use]
444 pub fn populated(version: i16) -> Self {
445 let mut m = Self::default();
446 if version >= 0 {
447 m.partition_index = 1i32;
448 }
449 if version >= 0 {
450 m.leader_id = 1i32;
451 }
452 if version >= 0 {
453 m.leader_epoch = 1i32;
454 }
455 if version == 0 {
456 m.preferred_successors = vec![1i32];
457 }
458 if version >= 1 {
459 m.preferred_candidates = vec![ReplicaInfo::populated(version)];
460 }
461 m
462 }
463}
464#[derive(Debug, Clone, PartialEq, Eq, Default)]
465pub struct ReplicaInfo {
466 pub candidate_id: i32,
467 pub candidate_directory_id: crate::primitives::uuid::Uuid,
468 pub unknown_tagged_fields: UnknownTaggedFields,
469}
470impl ReplicaInfo {
471 pub fn to_owned(&self) -> crate::owned::end_quorum_epoch_request::ReplicaInfo {
472 crate::owned::end_quorum_epoch_request::ReplicaInfo {
473 candidate_id: (self.candidate_id),
474 candidate_directory_id: (self.candidate_directory_id),
475 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
476 }
477 }
478}
479impl Encode for ReplicaInfo {
480 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
481 let flex = version >= 1;
482 if version >= 1 {
483 put_i32(buf, self.candidate_id);
484 }
485 if version >= 1 {
486 crate::primitives::uuid::put_uuid(buf, self.candidate_directory_id);
487 }
488 if flex {
489 let tagged = WriteTaggedFields::new();
490 tagged.write(buf, &self.unknown_tagged_fields);
491 }
492 Ok(())
493 }
494 fn encoded_len(&self, version: i16) -> usize {
495 let flex = version >= 1;
496 let mut n: usize = 0;
497 if version >= 1 {
498 n += 4;
499 }
500 if version >= 1 {
501 n += 16;
502 }
503 if flex {
504 let known_pairs: Vec<(u32, usize)> = Vec::new();
505 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
506 }
507 n
508 }
509}
510impl<'de> DecodeBorrow<'de> for ReplicaInfo {
511 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
512 let flex = version >= 1;
513 let mut out = Self::default();
514 if version >= 1 {
515 out.candidate_id = get_i32(buf)?;
516 }
517 if version >= 1 {
518 out.candidate_directory_id = crate::primitives::uuid::get_uuid(buf)?;
519 }
520 if flex {
521 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
522 }
523 Ok(out)
524 }
525}
526#[cfg(test)]
527impl ReplicaInfo {
528 #[must_use]
529 pub fn populated(version: i16) -> Self {
530 let mut m = Self::default();
531 if version >= 1 {
532 m.candidate_id = 1i32;
533 }
534 if version >= 1 {
535 m.candidate_directory_id = crate::primitives::uuid::Uuid([1u8; 16]);
536 }
537 m
538 }
539}
540#[derive(Debug, Clone, PartialEq, Eq, Default)]
541pub struct LeaderEndpoint<'a> {
542 pub name: &'a str,
543 pub host: &'a str,
544 pub port: u16,
545 pub unknown_tagged_fields: UnknownTaggedFields,
546}
547impl LeaderEndpoint<'_> {
548 pub fn to_owned(&self) -> crate::owned::end_quorum_epoch_request::LeaderEndpoint {
549 crate::owned::end_quorum_epoch_request::LeaderEndpoint {
550 name: (self.name).to_string(),
551 host: (self.host).to_string(),
552 port: (self.port),
553 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
554 }
555 }
556}
557impl Encode for LeaderEndpoint<'_> {
558 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
559 let flex = version >= 1;
560 if version >= 1 {
561 if flex {
562 put_compact_string(buf, self.name);
563 } else {
564 put_string(buf, self.name);
565 }
566 }
567 if version >= 1 {
568 if flex {
569 put_compact_string(buf, self.host);
570 } else {
571 put_string(buf, self.host);
572 }
573 }
574 if version >= 1 {
575 put_u16(buf, self.port);
576 }
577 if flex {
578 let tagged = WriteTaggedFields::new();
579 tagged.write(buf, &self.unknown_tagged_fields);
580 }
581 Ok(())
582 }
583 fn encoded_len(&self, version: i16) -> usize {
584 let flex = version >= 1;
585 let mut n: usize = 0;
586 if version >= 1 {
587 n += if flex {
588 compact_string_len(self.name)
589 } else {
590 string_len(self.name)
591 };
592 }
593 if version >= 1 {
594 n += if flex {
595 compact_string_len(self.host)
596 } else {
597 string_len(self.host)
598 };
599 }
600 if version >= 1 {
601 n += 2;
602 }
603 if flex {
604 let known_pairs: Vec<(u32, usize)> = Vec::new();
605 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
606 }
607 n
608 }
609}
610impl<'de> DecodeBorrow<'de> for LeaderEndpoint<'de> {
611 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
612 let flex = version >= 1;
613 let mut out = Self::default();
614 if version >= 1 {
615 out.name = if flex {
616 get_compact_string_borrowed(buf)?
617 } else {
618 get_string_borrowed(buf)?
619 };
620 }
621 if version >= 1 {
622 out.host = if flex {
623 get_compact_string_borrowed(buf)?
624 } else {
625 get_string_borrowed(buf)?
626 };
627 }
628 if version >= 1 {
629 out.port = get_u16(buf)?;
630 }
631 if flex {
632 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
633 }
634 Ok(out)
635 }
636}
637#[cfg(test)]
638impl LeaderEndpoint<'_> {
639 #[must_use]
640 pub fn populated(version: i16) -> Self {
641 let mut m = Self::default();
642 if version >= 1 {
643 m.name = "x";
644 }
645 if version >= 1 {
646 m.host = "x";
647 }
648 if version >= 1 {
649 m.port = 1u16;
650 }
651 m
652 }
653}