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