1#![allow(unused)]
6
7use std::borrow::Borrow;
8use std::collections::BTreeMap;
9
10use anyhow::{bail, Result};
11use bytes::Bytes;
12use uuid::Uuid;
13
14use crate::protocol::{
15 buf::{ByteBuf, ByteBufMut},
16 compute_unknown_tagged_fields_size, types, write_unknown_tagged_fields, Decodable, Decoder,
17 Encodable, Encoder, HeaderVersion, Message, StrBytes, VersionRange,
18};
19
20#[non_exhaustive]
22#[derive(Debug, Clone, PartialEq)]
23pub struct Cursor {
24 pub topic_name: super::TopicName,
28
29 pub partition_index: i32,
33
34 pub unknown_tagged_fields: BTreeMap<i32, Bytes>,
36}
37
38impl Cursor {
39 pub fn with_topic_name(mut self, value: super::TopicName) -> Self {
45 self.topic_name = value;
46 self
47 }
48 pub fn with_partition_index(mut self, value: i32) -> Self {
54 self.partition_index = value;
55 self
56 }
57 pub fn with_unknown_tagged_fields(mut self, value: BTreeMap<i32, Bytes>) -> Self {
59 self.unknown_tagged_fields = value;
60 self
61 }
62 pub fn with_unknown_tagged_field(mut self, key: i32, value: Bytes) -> Self {
64 self.unknown_tagged_fields.insert(key, value);
65 self
66 }
67}
68
69#[cfg(feature = "broker")]
70impl Encodable for Cursor {
71 fn encode<B: ByteBufMut>(&self, buf: &mut B, version: i16) -> Result<()> {
72 if version != 0 {
73 bail!("specified version not supported by this message type");
74 }
75 types::CompactString.encode(buf, &self.topic_name)?;
76 types::Int32.encode(buf, &self.partition_index)?;
77 let num_tagged_fields = self.unknown_tagged_fields.len();
78 if num_tagged_fields > std::u32::MAX as usize {
79 bail!(
80 "Too many tagged fields to encode ({} fields)",
81 num_tagged_fields
82 );
83 }
84 types::UnsignedVarInt.encode(buf, num_tagged_fields as u32)?;
85
86 write_unknown_tagged_fields(buf, 0.., &self.unknown_tagged_fields)?;
87 Ok(())
88 }
89 fn compute_size(&self, version: i16) -> Result<usize> {
90 let mut total_size = 0;
91 total_size += types::CompactString.compute_size(&self.topic_name)?;
92 total_size += types::Int32.compute_size(&self.partition_index)?;
93 let num_tagged_fields = self.unknown_tagged_fields.len();
94 if num_tagged_fields > std::u32::MAX as usize {
95 bail!(
96 "Too many tagged fields to encode ({} fields)",
97 num_tagged_fields
98 );
99 }
100 total_size += types::UnsignedVarInt.compute_size(num_tagged_fields as u32)?;
101
102 total_size += compute_unknown_tagged_fields_size(&self.unknown_tagged_fields)?;
103 Ok(total_size)
104 }
105}
106
107#[cfg(feature = "client")]
108impl Decodable for Cursor {
109 fn decode<B: ByteBuf>(buf: &mut B, version: i16) -> Result<Self> {
110 if version != 0 {
111 bail!("specified version not supported by this message type");
112 }
113 let topic_name = types::CompactString.decode(buf)?;
114 let partition_index = types::Int32.decode(buf)?;
115 let mut unknown_tagged_fields = BTreeMap::new();
116 let num_tagged_fields = types::UnsignedVarInt.decode(buf)?;
117 for _ in 0..num_tagged_fields {
118 let tag: u32 = types::UnsignedVarInt.decode(buf)?;
119 let size: u32 = types::UnsignedVarInt.decode(buf)?;
120 let unknown_value = buf.try_get_bytes(size as usize)?;
121 unknown_tagged_fields.insert(tag as i32, unknown_value);
122 }
123 Ok(Self {
124 topic_name,
125 partition_index,
126 unknown_tagged_fields,
127 })
128 }
129}
130
131impl Default for Cursor {
132 fn default() -> Self {
133 Self {
134 topic_name: Default::default(),
135 partition_index: 0,
136 unknown_tagged_fields: BTreeMap::new(),
137 }
138 }
139}
140
141impl Message for Cursor {
142 const VERSIONS: VersionRange = VersionRange { min: 0, max: 0 };
143 const DEPRECATED_VERSIONS: Option<VersionRange> = None;
144}
145
146#[non_exhaustive]
148#[derive(Debug, Clone, PartialEq)]
149pub struct DescribeTopicPartitionsResponse {
150 pub throttle_time_ms: i32,
154
155 pub topics: Vec<DescribeTopicPartitionsResponseTopic>,
159
160 pub next_cursor: Option<Cursor>,
164
165 pub unknown_tagged_fields: BTreeMap<i32, Bytes>,
167}
168
169impl DescribeTopicPartitionsResponse {
170 pub fn with_throttle_time_ms(mut self, value: i32) -> Self {
176 self.throttle_time_ms = value;
177 self
178 }
179 pub fn with_topics(mut self, value: Vec<DescribeTopicPartitionsResponseTopic>) -> Self {
185 self.topics = value;
186 self
187 }
188 pub fn with_next_cursor(mut self, value: Option<Cursor>) -> Self {
194 self.next_cursor = value;
195 self
196 }
197 pub fn with_unknown_tagged_fields(mut self, value: BTreeMap<i32, Bytes>) -> Self {
199 self.unknown_tagged_fields = value;
200 self
201 }
202 pub fn with_unknown_tagged_field(mut self, key: i32, value: Bytes) -> Self {
204 self.unknown_tagged_fields.insert(key, value);
205 self
206 }
207}
208
209#[cfg(feature = "broker")]
210impl Encodable for DescribeTopicPartitionsResponse {
211 fn encode<B: ByteBufMut>(&self, buf: &mut B, version: i16) -> Result<()> {
212 if version != 0 {
213 bail!("specified version not supported by this message type");
214 }
215 types::Int32.encode(buf, &self.throttle_time_ms)?;
216 types::CompactArray(types::Struct { version }).encode(buf, &self.topics)?;
217 types::OptionStruct { version }.encode(buf, &self.next_cursor)?;
218 let num_tagged_fields = self.unknown_tagged_fields.len();
219 if num_tagged_fields > std::u32::MAX as usize {
220 bail!(
221 "Too many tagged fields to encode ({} fields)",
222 num_tagged_fields
223 );
224 }
225 types::UnsignedVarInt.encode(buf, num_tagged_fields as u32)?;
226
227 write_unknown_tagged_fields(buf, 0.., &self.unknown_tagged_fields)?;
228 Ok(())
229 }
230 fn compute_size(&self, version: i16) -> Result<usize> {
231 let mut total_size = 0;
232 total_size += types::Int32.compute_size(&self.throttle_time_ms)?;
233 total_size += types::CompactArray(types::Struct { version }).compute_size(&self.topics)?;
234 total_size += types::OptionStruct { version }.compute_size(&self.next_cursor)?;
235 let num_tagged_fields = self.unknown_tagged_fields.len();
236 if num_tagged_fields > std::u32::MAX as usize {
237 bail!(
238 "Too many tagged fields to encode ({} fields)",
239 num_tagged_fields
240 );
241 }
242 total_size += types::UnsignedVarInt.compute_size(num_tagged_fields as u32)?;
243
244 total_size += compute_unknown_tagged_fields_size(&self.unknown_tagged_fields)?;
245 Ok(total_size)
246 }
247}
248
249#[cfg(feature = "client")]
250impl Decodable for DescribeTopicPartitionsResponse {
251 fn decode<B: ByteBuf>(buf: &mut B, version: i16) -> Result<Self> {
252 if version != 0 {
253 bail!("specified version not supported by this message type");
254 }
255 let throttle_time_ms = types::Int32.decode(buf)?;
256 let topics = types::CompactArray(types::Struct { version }).decode(buf)?;
257 let next_cursor = types::OptionStruct { version }.decode(buf)?;
258 let mut unknown_tagged_fields = BTreeMap::new();
259 let num_tagged_fields = types::UnsignedVarInt.decode(buf)?;
260 for _ in 0..num_tagged_fields {
261 let tag: u32 = types::UnsignedVarInt.decode(buf)?;
262 let size: u32 = types::UnsignedVarInt.decode(buf)?;
263 let unknown_value = buf.try_get_bytes(size as usize)?;
264 unknown_tagged_fields.insert(tag as i32, unknown_value);
265 }
266 Ok(Self {
267 throttle_time_ms,
268 topics,
269 next_cursor,
270 unknown_tagged_fields,
271 })
272 }
273}
274
275impl Default for DescribeTopicPartitionsResponse {
276 fn default() -> Self {
277 Self {
278 throttle_time_ms: 0,
279 topics: Default::default(),
280 next_cursor: None,
281 unknown_tagged_fields: BTreeMap::new(),
282 }
283 }
284}
285
286impl Message for DescribeTopicPartitionsResponse {
287 const VERSIONS: VersionRange = VersionRange { min: 0, max: 0 };
288 const DEPRECATED_VERSIONS: Option<VersionRange> = None;
289}
290
291#[non_exhaustive]
293#[derive(Debug, Clone, PartialEq)]
294pub struct DescribeTopicPartitionsResponsePartition {
295 pub error_code: i16,
299
300 pub partition_index: i32,
304
305 pub leader_id: super::BrokerId,
309
310 pub leader_epoch: i32,
314
315 pub replica_nodes: Vec<super::BrokerId>,
319
320 pub isr_nodes: Vec<super::BrokerId>,
324
325 pub eligible_leader_replicas: Option<Vec<super::BrokerId>>,
329
330 pub last_known_elr: Option<Vec<super::BrokerId>>,
334
335 pub offline_replicas: Vec<super::BrokerId>,
339
340 pub unknown_tagged_fields: BTreeMap<i32, Bytes>,
342}
343
344impl DescribeTopicPartitionsResponsePartition {
345 pub fn with_error_code(mut self, value: i16) -> Self {
351 self.error_code = value;
352 self
353 }
354 pub fn with_partition_index(mut self, value: i32) -> Self {
360 self.partition_index = value;
361 self
362 }
363 pub fn with_leader_id(mut self, value: super::BrokerId) -> Self {
369 self.leader_id = value;
370 self
371 }
372 pub fn with_leader_epoch(mut self, value: i32) -> Self {
378 self.leader_epoch = value;
379 self
380 }
381 pub fn with_replica_nodes(mut self, value: Vec<super::BrokerId>) -> Self {
387 self.replica_nodes = value;
388 self
389 }
390 pub fn with_isr_nodes(mut self, value: Vec<super::BrokerId>) -> Self {
396 self.isr_nodes = value;
397 self
398 }
399 pub fn with_eligible_leader_replicas(mut self, value: Option<Vec<super::BrokerId>>) -> Self {
405 self.eligible_leader_replicas = value;
406 self
407 }
408 pub fn with_last_known_elr(mut self, value: Option<Vec<super::BrokerId>>) -> Self {
414 self.last_known_elr = value;
415 self
416 }
417 pub fn with_offline_replicas(mut self, value: Vec<super::BrokerId>) -> Self {
423 self.offline_replicas = value;
424 self
425 }
426 pub fn with_unknown_tagged_fields(mut self, value: BTreeMap<i32, Bytes>) -> Self {
428 self.unknown_tagged_fields = value;
429 self
430 }
431 pub fn with_unknown_tagged_field(mut self, key: i32, value: Bytes) -> Self {
433 self.unknown_tagged_fields.insert(key, value);
434 self
435 }
436}
437
438#[cfg(feature = "broker")]
439impl Encodable for DescribeTopicPartitionsResponsePartition {
440 fn encode<B: ByteBufMut>(&self, buf: &mut B, version: i16) -> Result<()> {
441 if version != 0 {
442 bail!("specified version not supported by this message type");
443 }
444 types::Int16.encode(buf, &self.error_code)?;
445 types::Int32.encode(buf, &self.partition_index)?;
446 types::Int32.encode(buf, &self.leader_id)?;
447 types::Int32.encode(buf, &self.leader_epoch)?;
448 types::CompactArray(types::Int32).encode(buf, &self.replica_nodes)?;
449 types::CompactArray(types::Int32).encode(buf, &self.isr_nodes)?;
450 types::CompactArray(types::Int32).encode(buf, &self.eligible_leader_replicas)?;
451 types::CompactArray(types::Int32).encode(buf, &self.last_known_elr)?;
452 types::CompactArray(types::Int32).encode(buf, &self.offline_replicas)?;
453 let num_tagged_fields = self.unknown_tagged_fields.len();
454 if num_tagged_fields > std::u32::MAX as usize {
455 bail!(
456 "Too many tagged fields to encode ({} fields)",
457 num_tagged_fields
458 );
459 }
460 types::UnsignedVarInt.encode(buf, num_tagged_fields as u32)?;
461
462 write_unknown_tagged_fields(buf, 0.., &self.unknown_tagged_fields)?;
463 Ok(())
464 }
465 fn compute_size(&self, version: i16) -> Result<usize> {
466 let mut total_size = 0;
467 total_size += types::Int16.compute_size(&self.error_code)?;
468 total_size += types::Int32.compute_size(&self.partition_index)?;
469 total_size += types::Int32.compute_size(&self.leader_id)?;
470 total_size += types::Int32.compute_size(&self.leader_epoch)?;
471 total_size += types::CompactArray(types::Int32).compute_size(&self.replica_nodes)?;
472 total_size += types::CompactArray(types::Int32).compute_size(&self.isr_nodes)?;
473 total_size +=
474 types::CompactArray(types::Int32).compute_size(&self.eligible_leader_replicas)?;
475 total_size += types::CompactArray(types::Int32).compute_size(&self.last_known_elr)?;
476 total_size += types::CompactArray(types::Int32).compute_size(&self.offline_replicas)?;
477 let num_tagged_fields = self.unknown_tagged_fields.len();
478 if num_tagged_fields > std::u32::MAX as usize {
479 bail!(
480 "Too many tagged fields to encode ({} fields)",
481 num_tagged_fields
482 );
483 }
484 total_size += types::UnsignedVarInt.compute_size(num_tagged_fields as u32)?;
485
486 total_size += compute_unknown_tagged_fields_size(&self.unknown_tagged_fields)?;
487 Ok(total_size)
488 }
489}
490
491#[cfg(feature = "client")]
492impl Decodable for DescribeTopicPartitionsResponsePartition {
493 fn decode<B: ByteBuf>(buf: &mut B, version: i16) -> Result<Self> {
494 if version != 0 {
495 bail!("specified version not supported by this message type");
496 }
497 let error_code = types::Int16.decode(buf)?;
498 let partition_index = types::Int32.decode(buf)?;
499 let leader_id = types::Int32.decode(buf)?;
500 let leader_epoch = types::Int32.decode(buf)?;
501 let replica_nodes = types::CompactArray(types::Int32).decode(buf)?;
502 let isr_nodes = types::CompactArray(types::Int32).decode(buf)?;
503 let eligible_leader_replicas = types::CompactArray(types::Int32).decode(buf)?;
504 let last_known_elr = types::CompactArray(types::Int32).decode(buf)?;
505 let offline_replicas = types::CompactArray(types::Int32).decode(buf)?;
506 let mut unknown_tagged_fields = BTreeMap::new();
507 let num_tagged_fields = types::UnsignedVarInt.decode(buf)?;
508 for _ in 0..num_tagged_fields {
509 let tag: u32 = types::UnsignedVarInt.decode(buf)?;
510 let size: u32 = types::UnsignedVarInt.decode(buf)?;
511 let unknown_value = buf.try_get_bytes(size as usize)?;
512 unknown_tagged_fields.insert(tag as i32, unknown_value);
513 }
514 Ok(Self {
515 error_code,
516 partition_index,
517 leader_id,
518 leader_epoch,
519 replica_nodes,
520 isr_nodes,
521 eligible_leader_replicas,
522 last_known_elr,
523 offline_replicas,
524 unknown_tagged_fields,
525 })
526 }
527}
528
529impl Default for DescribeTopicPartitionsResponsePartition {
530 fn default() -> Self {
531 Self {
532 error_code: 0,
533 partition_index: 0,
534 leader_id: (0).into(),
535 leader_epoch: -1,
536 replica_nodes: Default::default(),
537 isr_nodes: Default::default(),
538 eligible_leader_replicas: None,
539 last_known_elr: None,
540 offline_replicas: Default::default(),
541 unknown_tagged_fields: BTreeMap::new(),
542 }
543 }
544}
545
546impl Message for DescribeTopicPartitionsResponsePartition {
547 const VERSIONS: VersionRange = VersionRange { min: 0, max: 0 };
548 const DEPRECATED_VERSIONS: Option<VersionRange> = None;
549}
550
551#[non_exhaustive]
553#[derive(Debug, Clone, PartialEq)]
554pub struct DescribeTopicPartitionsResponseTopic {
555 pub error_code: i16,
559
560 pub name: Option<super::TopicName>,
564
565 pub topic_id: Uuid,
569
570 pub is_internal: bool,
574
575 pub partitions: Vec<DescribeTopicPartitionsResponsePartition>,
579
580 pub topic_authorized_operations: i32,
584
585 pub unknown_tagged_fields: BTreeMap<i32, Bytes>,
587}
588
589impl DescribeTopicPartitionsResponseTopic {
590 pub fn with_error_code(mut self, value: i16) -> Self {
596 self.error_code = value;
597 self
598 }
599 pub fn with_name(mut self, value: Option<super::TopicName>) -> Self {
605 self.name = value;
606 self
607 }
608 pub fn with_topic_id(mut self, value: Uuid) -> Self {
614 self.topic_id = value;
615 self
616 }
617 pub fn with_is_internal(mut self, value: bool) -> Self {
623 self.is_internal = value;
624 self
625 }
626 pub fn with_partitions(mut self, value: Vec<DescribeTopicPartitionsResponsePartition>) -> Self {
632 self.partitions = value;
633 self
634 }
635 pub fn with_topic_authorized_operations(mut self, value: i32) -> Self {
641 self.topic_authorized_operations = value;
642 self
643 }
644 pub fn with_unknown_tagged_fields(mut self, value: BTreeMap<i32, Bytes>) -> Self {
646 self.unknown_tagged_fields = value;
647 self
648 }
649 pub fn with_unknown_tagged_field(mut self, key: i32, value: Bytes) -> Self {
651 self.unknown_tagged_fields.insert(key, value);
652 self
653 }
654}
655
656#[cfg(feature = "broker")]
657impl Encodable for DescribeTopicPartitionsResponseTopic {
658 fn encode<B: ByteBufMut>(&self, buf: &mut B, version: i16) -> Result<()> {
659 if version != 0 {
660 bail!("specified version not supported by this message type");
661 }
662 types::Int16.encode(buf, &self.error_code)?;
663 types::CompactString.encode(buf, &self.name)?;
664 types::Uuid.encode(buf, &self.topic_id)?;
665 types::Boolean.encode(buf, &self.is_internal)?;
666 types::CompactArray(types::Struct { version }).encode(buf, &self.partitions)?;
667 types::Int32.encode(buf, &self.topic_authorized_operations)?;
668 let num_tagged_fields = self.unknown_tagged_fields.len();
669 if num_tagged_fields > std::u32::MAX as usize {
670 bail!(
671 "Too many tagged fields to encode ({} fields)",
672 num_tagged_fields
673 );
674 }
675 types::UnsignedVarInt.encode(buf, num_tagged_fields as u32)?;
676
677 write_unknown_tagged_fields(buf, 0.., &self.unknown_tagged_fields)?;
678 Ok(())
679 }
680 fn compute_size(&self, version: i16) -> Result<usize> {
681 let mut total_size = 0;
682 total_size += types::Int16.compute_size(&self.error_code)?;
683 total_size += types::CompactString.compute_size(&self.name)?;
684 total_size += types::Uuid.compute_size(&self.topic_id)?;
685 total_size += types::Boolean.compute_size(&self.is_internal)?;
686 total_size +=
687 types::CompactArray(types::Struct { version }).compute_size(&self.partitions)?;
688 total_size += types::Int32.compute_size(&self.topic_authorized_operations)?;
689 let num_tagged_fields = self.unknown_tagged_fields.len();
690 if num_tagged_fields > std::u32::MAX as usize {
691 bail!(
692 "Too many tagged fields to encode ({} fields)",
693 num_tagged_fields
694 );
695 }
696 total_size += types::UnsignedVarInt.compute_size(num_tagged_fields as u32)?;
697
698 total_size += compute_unknown_tagged_fields_size(&self.unknown_tagged_fields)?;
699 Ok(total_size)
700 }
701}
702
703#[cfg(feature = "client")]
704impl Decodable for DescribeTopicPartitionsResponseTopic {
705 fn decode<B: ByteBuf>(buf: &mut B, version: i16) -> Result<Self> {
706 if version != 0 {
707 bail!("specified version not supported by this message type");
708 }
709 let error_code = types::Int16.decode(buf)?;
710 let name = types::CompactString.decode(buf)?;
711 let topic_id = types::Uuid.decode(buf)?;
712 let is_internal = types::Boolean.decode(buf)?;
713 let partitions = types::CompactArray(types::Struct { version }).decode(buf)?;
714 let topic_authorized_operations = types::Int32.decode(buf)?;
715 let mut unknown_tagged_fields = BTreeMap::new();
716 let num_tagged_fields = types::UnsignedVarInt.decode(buf)?;
717 for _ in 0..num_tagged_fields {
718 let tag: u32 = types::UnsignedVarInt.decode(buf)?;
719 let size: u32 = types::UnsignedVarInt.decode(buf)?;
720 let unknown_value = buf.try_get_bytes(size as usize)?;
721 unknown_tagged_fields.insert(tag as i32, unknown_value);
722 }
723 Ok(Self {
724 error_code,
725 name,
726 topic_id,
727 is_internal,
728 partitions,
729 topic_authorized_operations,
730 unknown_tagged_fields,
731 })
732 }
733}
734
735impl Default for DescribeTopicPartitionsResponseTopic {
736 fn default() -> Self {
737 Self {
738 error_code: 0,
739 name: Some(Default::default()),
740 topic_id: Uuid::nil(),
741 is_internal: false,
742 partitions: Default::default(),
743 topic_authorized_operations: -2147483648,
744 unknown_tagged_fields: BTreeMap::new(),
745 }
746 }
747}
748
749impl Message for DescribeTopicPartitionsResponseTopic {
750 const VERSIONS: VersionRange = VersionRange { min: 0, max: 0 };
751 const DEPRECATED_VERSIONS: Option<VersionRange> = None;
752}
753
754impl HeaderVersion for DescribeTopicPartitionsResponse {
755 fn header_version(version: i16) -> i16 {
756 1
757 }
758}