1use bytes::{Buf, BufMut};
4
5use crate::{
6 Decode, Encode, ProtocolError, UnknownTaggedFields,
7 primitives::{
8 fixed::{get_bool, get_i16, get_i32, put_bool, put_i16, put_i32},
9 string_bytes::{
10 compact_nullable_string_len, compact_string_len, get_compact_nullable_string_owned,
11 get_compact_string_owned, get_nullable_string_owned, get_string_owned,
12 nullable_string_len, put_compact_nullable_string, put_compact_string,
13 put_nullable_string, put_string, string_len,
14 },
15 },
16 tagged_fields::{WriteTaggedFields, read_tagged_fields, tagged_fields_len},
17};
18pub const API_KEY: i16 = 3;
19pub const MIN_VERSION: i16 = 0;
20pub const MAX_VERSION: i16 = 13;
21pub const FLEXIBLE_MIN: i16 = 9;
22#[inline]
23#[must_use]
24pub fn is_flexible(version: i16) -> bool {
25 version >= FLEXIBLE_MIN
26}
27#[derive(Debug, Clone, PartialEq, Eq)]
28pub struct MetadataResponse {
29 pub throttle_time_ms: i32,
30 pub brokers: Vec<MetadataResponseBroker>,
31 pub cluster_id: Option<String>,
32 pub controller_id: i32,
33 pub topics: Vec<MetadataResponseTopic>,
34 pub cluster_authorized_operations: i32,
35 pub error_code: i16,
36 pub unknown_tagged_fields: UnknownTaggedFields,
37}
38impl Default for MetadataResponse {
39 fn default() -> Self {
40 Self {
41 throttle_time_ms: 0i32,
42 brokers: Vec::new(),
43 cluster_id: None,
44 controller_id: -1i32,
45 topics: Vec::new(),
46 cluster_authorized_operations: -2_147_483_648i32,
47 error_code: 0i16,
48 unknown_tagged_fields: UnknownTaggedFields::default(),
49 }
50 }
51}
52impl Encode for MetadataResponse {
53 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
54 if !(MIN_VERSION..=MAX_VERSION).contains(&version) {
55 return Err(ProtocolError::UnsupportedVersion {
56 api_key: API_KEY,
57 version,
58 });
59 }
60 let flex = is_flexible(version);
61 if version >= 3 {
62 put_i32(buf, self.throttle_time_ms);
63 }
64 if version >= 0 {
65 {
66 crate::primitives::array::put_array_len(buf, (self.brokers).len(), flex);
67 for it in &self.brokers {
68 it.encode(buf, version)?;
69 }
70 };
71 }
72 if version >= 2 {
73 if flex {
74 let () = put_compact_nullable_string(buf, self.cluster_id.as_deref());
75 } else {
76 let () = put_nullable_string(buf, self.cluster_id.as_deref());
77 }
78 }
79 if version >= 1 {
80 put_i32(buf, self.controller_id);
81 }
82 if version >= 0 {
83 {
84 crate::primitives::array::put_array_len(buf, (self.topics).len(), flex);
85 for it in &self.topics {
86 it.encode(buf, version)?;
87 }
88 };
89 }
90 if (8..=10).contains(&version) {
91 put_i32(buf, self.cluster_authorized_operations);
92 }
93 if version >= 13 {
94 put_i16(buf, self.error_code);
95 }
96 if flex {
97 let tagged = WriteTaggedFields::new();
98 tagged.write(buf, &self.unknown_tagged_fields);
99 }
100 Ok(())
101 }
102 fn encoded_len(&self, version: i16) -> usize {
103 let flex = is_flexible(version);
104 let mut n: usize = 0;
105 if version >= 3 {
106 n += 4;
107 }
108 if version >= 0 {
109 n += {
110 let prefix =
111 crate::primitives::array::array_len_prefix_len((self.brokers).len(), flex);
112 let body: usize = (self.brokers)
113 .iter()
114 .map(|it| it.encoded_len(version))
115 .sum();
116 prefix + body
117 };
118 }
119 if version >= 2 {
120 n += if flex {
121 compact_nullable_string_len(self.cluster_id.as_deref())
122 } else {
123 nullable_string_len(self.cluster_id.as_deref())
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.topics).len(), flex);
133 let body: usize = (self.topics).iter().map(|it| it.encoded_len(version)).sum();
134 prefix + body
135 };
136 }
137 if (8..=10).contains(&version) {
138 n += 4;
139 }
140 if version >= 13 {
141 n += 2;
142 }
143 if flex {
144 let known_pairs: Vec<(u32, usize)> = Vec::new();
145 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
146 }
147 n
148 }
149}
150impl Decode<'_> for MetadataResponse {
151 fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
152 if !(MIN_VERSION..=MAX_VERSION).contains(&version) {
153 return Err(ProtocolError::UnsupportedVersion {
154 api_key: API_KEY,
155 version,
156 });
157 }
158 let flex = is_flexible(version);
159 let mut out = Self::default();
160 if version >= 3 {
161 out.throttle_time_ms = get_i32(buf)?;
162 }
163 if version >= 0 {
164 out.brokers = {
165 let n = crate::primitives::array::get_array_len(buf, flex)?;
166 let mut v = Vec::with_capacity(n);
167 for _ in 0..n {
168 v.push(MetadataResponseBroker::decode(buf, version)?);
169 }
170 v
171 };
172 }
173 if version >= 2 {
174 out.cluster_id = if flex {
175 get_compact_nullable_string_owned(buf)?
176 } else {
177 get_nullable_string_owned(buf)?
178 };
179 }
180 if version >= 1 {
181 out.controller_id = get_i32(buf)?;
182 }
183 if version >= 0 {
184 out.topics = {
185 let n = crate::primitives::array::get_array_len(buf, flex)?;
186 let mut v = Vec::with_capacity(n);
187 for _ in 0..n {
188 v.push(MetadataResponseTopic::decode(buf, version)?);
189 }
190 v
191 };
192 }
193 if (8..=10).contains(&version) {
194 out.cluster_authorized_operations = get_i32(buf)?;
195 }
196 if version >= 13 {
197 out.error_code = get_i16(buf)?;
198 }
199 if flex {
200 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
201 }
202 Ok(out)
203 }
204}
205#[cfg(test)]
206impl MetadataResponse {
207 #[must_use]
208 pub fn populated(version: i16) -> Self {
209 let mut m = Self::default();
210 if version >= 3 {
211 m.throttle_time_ms = 1i32;
212 }
213 if version >= 0 {
214 m.brokers = vec![MetadataResponseBroker::populated(version)];
215 }
216 if version >= 2 {
217 m.cluster_id = Some("x".to_string());
218 }
219 if version >= 1 {
220 m.controller_id = 1i32;
221 }
222 if version >= 0 {
223 m.topics = vec![MetadataResponseTopic::populated(version)];
224 }
225 if (8..=10).contains(&version) {
226 m.cluster_authorized_operations = 1i32;
227 }
228 if version >= 13 {
229 m.error_code = 1i16;
230 }
231 m
232 }
233}
234#[derive(Debug, Clone, PartialEq, Eq, Default)]
235pub struct MetadataResponseBroker {
236 pub node_id: i32,
237 pub host: String,
238 pub port: i32,
239 pub rack: Option<String>,
240 pub unknown_tagged_fields: UnknownTaggedFields,
241}
242impl Encode for MetadataResponseBroker {
243 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
244 let flex = version >= 9;
245 if version >= 0 {
246 put_i32(buf, self.node_id);
247 }
248 if version >= 0 {
249 if flex {
250 let () = put_compact_string(buf, &self.host);
251 } else {
252 let () = put_string(buf, &self.host);
253 }
254 }
255 if version >= 0 {
256 put_i32(buf, self.port);
257 }
258 if version >= 1 {
259 if flex {
260 let () = put_compact_nullable_string(buf, self.rack.as_deref());
261 } else {
262 let () = put_nullable_string(buf, self.rack.as_deref());
263 }
264 }
265 if flex {
266 let tagged = WriteTaggedFields::new();
267 tagged.write(buf, &self.unknown_tagged_fields);
268 }
269 Ok(())
270 }
271 fn encoded_len(&self, version: i16) -> usize {
272 let flex = version >= 9;
273 let mut n: usize = 0;
274 if version >= 0 {
275 n += 4;
276 }
277 if version >= 0 {
278 n += if flex {
279 compact_string_len(&self.host)
280 } else {
281 string_len(&self.host)
282 };
283 }
284 if version >= 0 {
285 n += 4;
286 }
287 if version >= 1 {
288 n += if flex {
289 compact_nullable_string_len(self.rack.as_deref())
290 } else {
291 nullable_string_len(self.rack.as_deref())
292 };
293 }
294 if flex {
295 let known_pairs: Vec<(u32, usize)> = Vec::new();
296 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
297 }
298 n
299 }
300}
301impl Decode<'_> for MetadataResponseBroker {
302 fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
303 let flex = version >= 9;
304 let mut out = Self::default();
305 if version >= 0 {
306 out.node_id = get_i32(buf)?;
307 }
308 if version >= 0 {
309 out.host = if flex {
310 get_compact_string_owned(buf)?
311 } else {
312 get_string_owned(buf)?
313 };
314 }
315 if version >= 0 {
316 out.port = get_i32(buf)?;
317 }
318 if version >= 1 {
319 out.rack = if flex {
320 get_compact_nullable_string_owned(buf)?
321 } else {
322 get_nullable_string_owned(buf)?
323 };
324 }
325 if flex {
326 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
327 }
328 Ok(out)
329 }
330}
331#[cfg(test)]
332impl MetadataResponseBroker {
333 #[must_use]
334 pub fn populated(version: i16) -> Self {
335 let mut m = Self::default();
336 if version >= 0 {
337 m.node_id = 1i32;
338 }
339 if version >= 0 {
340 m.host = "x".to_string();
341 }
342 if version >= 0 {
343 m.port = 1i32;
344 }
345 if version >= 1 {
346 m.rack = Some("x".to_string());
347 }
348 m
349 }
350}
351#[derive(Debug, Clone, PartialEq, Eq)]
352pub struct MetadataResponseTopic {
353 pub error_code: i16,
354 pub name: Option<String>,
355 pub topic_id: crate::primitives::uuid::Uuid,
356 pub is_internal: bool,
357 pub partitions: Vec<MetadataResponsePartition>,
358 pub topic_authorized_operations: i32,
359 pub unknown_tagged_fields: UnknownTaggedFields,
360}
361impl Default for MetadataResponseTopic {
362 fn default() -> Self {
363 Self {
364 error_code: 0i16,
365 name: Some(String::new()),
366 topic_id: crate::primitives::uuid::Uuid::default(),
367 is_internal: false,
368 partitions: Vec::new(),
369 topic_authorized_operations: -2_147_483_648i32,
370 unknown_tagged_fields: UnknownTaggedFields::default(),
371 }
372 }
373}
374impl Encode for MetadataResponseTopic {
375 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
376 let flex = version >= 9;
377 if version >= 0 {
378 put_i16(buf, self.error_code);
379 }
380 if version >= 0 {
381 if version >= 12 {
382 if flex {
383 let () = put_compact_nullable_string(buf, self.name.as_deref());
384 } else {
385 let () = put_nullable_string(buf, self.name.as_deref());
386 }
387 } else {
388 if flex {
389 let () = put_compact_string(buf, (self.name).as_deref().unwrap_or(""));
390 } else {
391 let () = put_string(buf, (self.name).as_deref().unwrap_or(""));
392 }
393 }
394 }
395 if version >= 10 {
396 crate::primitives::uuid::put_uuid(buf, self.topic_id);
397 }
398 if version >= 1 {
399 put_bool(buf, self.is_internal);
400 }
401 if version >= 0 {
402 {
403 crate::primitives::array::put_array_len(buf, (self.partitions).len(), flex);
404 for it in &self.partitions {
405 it.encode(buf, version)?;
406 }
407 };
408 }
409 if version >= 8 {
410 put_i32(buf, self.topic_authorized_operations);
411 }
412 if flex {
413 let tagged = WriteTaggedFields::new();
414 tagged.write(buf, &self.unknown_tagged_fields);
415 }
416 Ok(())
417 }
418 fn encoded_len(&self, version: i16) -> usize {
419 let flex = version >= 9;
420 let mut n: usize = 0;
421 if version >= 0 {
422 n += 2;
423 }
424 if version >= 0 {
425 n += if version >= 12 {
426 if flex {
427 compact_nullable_string_len(self.name.as_deref())
428 } else {
429 nullable_string_len(self.name.as_deref())
430 }
431 } else {
432 if flex {
433 compact_string_len((self.name).as_deref().unwrap_or(""))
434 } else {
435 string_len((self.name).as_deref().unwrap_or(""))
436 }
437 };
438 }
439 if version >= 10 {
440 n += 16;
441 }
442 if version >= 1 {
443 n += 1;
444 }
445 if version >= 0 {
446 n += {
447 let prefix =
448 crate::primitives::array::array_len_prefix_len((self.partitions).len(), flex);
449 let body: usize = (self.partitions)
450 .iter()
451 .map(|it| it.encoded_len(version))
452 .sum();
453 prefix + body
454 };
455 }
456 if version >= 8 {
457 n += 4;
458 }
459 if flex {
460 let known_pairs: Vec<(u32, usize)> = Vec::new();
461 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
462 }
463 n
464 }
465}
466impl Decode<'_> for MetadataResponseTopic {
467 fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
468 let flex = version >= 9;
469 let mut out = Self::default();
470 if version >= 0 {
471 out.error_code = get_i16(buf)?;
472 }
473 if version >= 0 {
474 out.name = if version >= 12 {
475 if flex {
476 get_compact_nullable_string_owned(buf)?
477 } else {
478 get_nullable_string_owned(buf)?
479 }
480 } else {
481 Some(if flex {
482 get_compact_string_owned(buf)?
483 } else {
484 get_string_owned(buf)?
485 })
486 };
487 }
488 if version >= 10 {
489 out.topic_id = crate::primitives::uuid::get_uuid(buf)?;
490 }
491 if version >= 1 {
492 out.is_internal = get_bool(buf)?;
493 }
494 if version >= 0 {
495 out.partitions = {
496 let n = crate::primitives::array::get_array_len(buf, flex)?;
497 let mut v = Vec::with_capacity(n);
498 for _ in 0..n {
499 v.push(MetadataResponsePartition::decode(buf, version)?);
500 }
501 v
502 };
503 }
504 if version >= 8 {
505 out.topic_authorized_operations = get_i32(buf)?;
506 }
507 if flex {
508 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
509 }
510 Ok(out)
511 }
512}
513#[cfg(test)]
514impl MetadataResponseTopic {
515 #[must_use]
516 pub fn populated(version: i16) -> Self {
517 let mut m = Self::default();
518 if version >= 0 {
519 m.error_code = 1i16;
520 }
521 if version >= 0 {
522 m.name = Some("x".to_string());
523 }
524 if version >= 10 {
525 m.topic_id = crate::primitives::uuid::Uuid([1u8; 16]);
526 }
527 if version >= 1 {
528 m.is_internal = true;
529 }
530 if version >= 0 {
531 m.partitions = vec![MetadataResponsePartition::populated(version)];
532 }
533 if version >= 8 {
534 m.topic_authorized_operations = 1i32;
535 }
536 m
537 }
538}
539#[derive(Debug, Clone, PartialEq, Eq)]
540pub struct MetadataResponsePartition {
541 pub error_code: i16,
542 pub partition_index: i32,
543 pub leader_id: i32,
544 pub leader_epoch: i32,
545 pub replica_nodes: Vec<i32>,
546 pub isr_nodes: Vec<i32>,
547 pub offline_replicas: Vec<i32>,
548 pub unknown_tagged_fields: UnknownTaggedFields,
549}
550impl Default for MetadataResponsePartition {
551 fn default() -> Self {
552 Self {
553 error_code: 0i16,
554 partition_index: 0i32,
555 leader_id: 0i32,
556 leader_epoch: -1i32,
557 replica_nodes: Vec::new(),
558 isr_nodes: Vec::new(),
559 offline_replicas: Vec::new(),
560 unknown_tagged_fields: UnknownTaggedFields::default(),
561 }
562 }
563}
564impl Encode for MetadataResponsePartition {
565 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
566 let flex = version >= 9;
567 if version >= 0 {
568 put_i16(buf, self.error_code);
569 }
570 if version >= 0 {
571 put_i32(buf, self.partition_index);
572 }
573 if version >= 0 {
574 put_i32(buf, self.leader_id);
575 }
576 if version >= 7 {
577 put_i32(buf, self.leader_epoch);
578 }
579 if version >= 0 {
580 {
581 crate::primitives::array::put_array_len(buf, (self.replica_nodes).len(), flex);
582 for it in &self.replica_nodes {
583 put_i32(buf, *it);
584 }
585 };
586 }
587 if version >= 0 {
588 {
589 crate::primitives::array::put_array_len(buf, (self.isr_nodes).len(), flex);
590 for it in &self.isr_nodes {
591 put_i32(buf, *it);
592 }
593 };
594 }
595 if version >= 5 {
596 {
597 crate::primitives::array::put_array_len(buf, (self.offline_replicas).len(), flex);
598 for it in &self.offline_replicas {
599 put_i32(buf, *it);
600 }
601 };
602 }
603 if flex {
604 let tagged = WriteTaggedFields::new();
605 tagged.write(buf, &self.unknown_tagged_fields);
606 }
607 Ok(())
608 }
609 fn encoded_len(&self, version: i16) -> usize {
610 let flex = version >= 9;
611 let mut n: usize = 0;
612 if version >= 0 {
613 n += 2;
614 }
615 if version >= 0 {
616 n += 4;
617 }
618 if version >= 0 {
619 n += 4;
620 }
621 if version >= 7 {
622 n += 4;
623 }
624 if version >= 0 {
625 n += {
626 let prefix = crate::primitives::array::array_len_prefix_len(
627 (self.replica_nodes).len(),
628 flex,
629 );
630 let body: usize = (self.replica_nodes).iter().map(|_| 4).sum();
631 prefix + body
632 };
633 }
634 if version >= 0 {
635 n += {
636 let prefix =
637 crate::primitives::array::array_len_prefix_len((self.isr_nodes).len(), flex);
638 let body: usize = (self.isr_nodes).iter().map(|_| 4).sum();
639 prefix + body
640 };
641 }
642 if version >= 5 {
643 n += {
644 let prefix = crate::primitives::array::array_len_prefix_len(
645 (self.offline_replicas).len(),
646 flex,
647 );
648 let body: usize = (self.offline_replicas).iter().map(|_| 4).sum();
649 prefix + body
650 };
651 }
652 if flex {
653 let known_pairs: Vec<(u32, usize)> = Vec::new();
654 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
655 }
656 n
657 }
658}
659impl Decode<'_> for MetadataResponsePartition {
660 fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
661 let flex = version >= 9;
662 let mut out = Self::default();
663 if version >= 0 {
664 out.error_code = get_i16(buf)?;
665 }
666 if version >= 0 {
667 out.partition_index = get_i32(buf)?;
668 }
669 if version >= 0 {
670 out.leader_id = get_i32(buf)?;
671 }
672 if version >= 7 {
673 out.leader_epoch = get_i32(buf)?;
674 }
675 if version >= 0 {
676 out.replica_nodes = {
677 let n = crate::primitives::array::get_array_len(buf, flex)?;
678 let mut v = Vec::with_capacity(n);
679 for _ in 0..n {
680 v.push(get_i32(buf)?);
681 }
682 v
683 };
684 }
685 if version >= 0 {
686 out.isr_nodes = {
687 let n = crate::primitives::array::get_array_len(buf, flex)?;
688 let mut v = Vec::with_capacity(n);
689 for _ in 0..n {
690 v.push(get_i32(buf)?);
691 }
692 v
693 };
694 }
695 if version >= 5 {
696 out.offline_replicas = {
697 let n = crate::primitives::array::get_array_len(buf, flex)?;
698 let mut v = Vec::with_capacity(n);
699 for _ in 0..n {
700 v.push(get_i32(buf)?);
701 }
702 v
703 };
704 }
705 if flex {
706 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
707 }
708 Ok(out)
709 }
710}
711#[cfg(test)]
712impl MetadataResponsePartition {
713 #[must_use]
714 pub fn populated(version: i16) -> Self {
715 let mut m = Self::default();
716 if version >= 0 {
717 m.error_code = 1i16;
718 }
719 if version >= 0 {
720 m.partition_index = 1i32;
721 }
722 if version >= 0 {
723 m.leader_id = 1i32;
724 }
725 if version >= 7 {
726 m.leader_epoch = 1i32;
727 }
728 if version >= 0 {
729 m.replica_nodes = vec![1i32];
730 }
731 if version >= 0 {
732 m.isr_nodes = vec![1i32];
733 }
734 if version >= 5 {
735 m.offline_replicas = vec![1i32];
736 }
737 m
738 }
739}
740#[must_use]
743#[allow(unused_comparisons)]
744pub fn default_json(version: i16) -> ::serde_json::Value {
745 let mut obj = ::serde_json::Map::new();
746 if version >= 3 {
747 obj.insert("throttleTimeMs".to_string(), ::serde_json::json!(0));
748 }
749 obj.insert("brokers".to_string(), ::serde_json::Value::Array(vec![]));
750 if version >= 2 {
751 obj.insert("clusterId".to_string(), ::serde_json::Value::Null);
752 }
753 if version >= 1 {
754 obj.insert("controllerId".to_string(), ::serde_json::json!(-1));
755 }
756 obj.insert("topics".to_string(), ::serde_json::Value::Array(vec![]));
757 if (8..=10).contains(&version) {
758 obj.insert(
759 "clusterAuthorizedOperations".to_string(),
760 ::serde_json::json!(-2_147_483_648),
761 );
762 }
763 if version >= 13 {
764 obj.insert("errorCode".to_string(), ::serde_json::json!(0));
765 }
766 ::serde_json::Value::Object(obj)
767}