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