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