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