1use bytes::BufMut;
4
5use crate::primitives::fixed::{get_bool, get_i32, get_i64, get_i8, put_bool, put_i32, put_i64, put_i8};
6use crate::primitives::string_bytes::{
7 compact_nullable_string_len, compact_string_len, nullable_string_len,
8 put_compact_nullable_string, put_compact_string, put_nullable_string, put_string,
9 string_len,
10};
11use crate::primitives::string_bytes_borrowed::{
12 get_compact_nullable_string_borrowed, get_compact_string_borrowed,
13 get_nullable_string_borrowed, get_string_borrowed,
14};
15use crate::tagged_fields::{read_tagged_fields, tagged_fields_len, WriteTaggedFields};
16use crate::{DecodeBorrow, Encode, ProtocolError, UnknownTaggedFields};
17
18pub const API_KEY: i16 = 79;
19pub const MIN_VERSION: i16 = 1;
20pub const MAX_VERSION: i16 = 2;
21pub const FLEXIBLE_MIN: i16 = 0;
22
23#[inline]
24fn is_flexible(version: i16) -> bool { version >= FLEXIBLE_MIN }
25
26#[derive(Debug, Clone, PartialEq, Eq)]
27pub struct ShareAcknowledgeRequest<'a> {
28 pub group_id: Option<&'a str>,
29 pub member_id: Option<&'a str>,
30 pub share_session_epoch: i32,
31 pub is_renew_ack: bool,
32 pub topics: Vec<AcknowledgeTopic>,
33 pub unknown_tagged_fields: UnknownTaggedFields,
34}
35
36impl<'a> Default for ShareAcknowledgeRequest<'a> {
37 fn default() -> Self {
38 Self {
39 group_id: None,
40 member_id: None,
41 share_session_epoch: 0i32,
42 is_renew_ack: false,
43 topics: Vec::new(),
44 unknown_tagged_fields: Default::default(),
45 }
46 }
47}
48
49impl<'a> ShareAcknowledgeRequest<'a> {
50 pub fn to_owned(&self) -> crate::owned::share_acknowledge_request::ShareAcknowledgeRequest {
51 crate::owned::share_acknowledge_request::ShareAcknowledgeRequest {
52 group_id: (self.group_id).map(|s| s.to_string()),
53 member_id: (self.member_id).map(|s| s.to_string()),
54 share_session_epoch: (self.share_session_epoch),
55 is_renew_ack: (self.is_renew_ack),
56 topics: (self.topics).iter().map(|it| it.to_owned()).collect(),
57 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
58 }
59 }
60}
61
62impl<'a> Encode for ShareAcknowledgeRequest<'a> {
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 { api_key: API_KEY, version });
66 }
67 let flex = is_flexible(version);
68 if version >= 0 { if flex { put_compact_nullable_string(buf, self.group_id) } else { put_nullable_string(buf, self.group_id) } }
69 if version >= 0 { if flex { put_compact_nullable_string(buf, self.member_id) } else { put_nullable_string(buf, self.member_id) } }
70 if version >= 0 { put_i32(buf, self.share_session_epoch) }
71 if version >= 2 { put_bool(buf, self.is_renew_ack) }
72 if version >= 0 { { crate::primitives::array::put_array_len(buf, (self.topics).len(), flex); for it in &self.topics { it.encode(buf, version)?; } } }
73 if flex {
74 let tagged = WriteTaggedFields::new();
75 tagged.write(buf, &self.unknown_tagged_fields);
76 }
77 Ok(())
78 }
79 fn encoded_len(&self, version: i16) -> usize {
80 let flex = is_flexible(version);
81 let mut n: usize = 0;
82 if version >= 0 { n += if flex { compact_nullable_string_len(self.group_id) } else { nullable_string_len(self.group_id) }; }
83 if version >= 0 { n += if flex { compact_nullable_string_len(self.member_id) } else { nullable_string_len(self.member_id) }; }
84 if version >= 0 { n += 4; }
85 if version >= 2 { n += 1; }
86 if version >= 0 { n += { let prefix = crate::primitives::array::array_len_prefix_len((self.topics).len(), flex); let body: usize = (self.topics).iter().map(|it| it.encoded_len(version)).sum(); prefix + body }; }
87 if flex {
88 let known_pairs: Vec<(u32, usize)> = Vec::new();
89 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
90 }
91 n
92 }
93}
94
95impl<'de> DecodeBorrow<'de> for ShareAcknowledgeRequest<'de> {
96 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
97 if !(MIN_VERSION..=MAX_VERSION).contains(&version) {
98 return Err(ProtocolError::UnsupportedVersion { api_key: API_KEY, version });
99 }
100 let flex = is_flexible(version);
101 let mut out = Self::default();
102 if version >= 0 { out.group_id = if flex { get_compact_nullable_string_borrowed(buf)? } else { get_nullable_string_borrowed(buf)? }; }
103 if version >= 0 { out.member_id = if flex { get_compact_nullable_string_borrowed(buf)? } else { get_nullable_string_borrowed(buf)? }; }
104 if version >= 0 { out.share_session_epoch = get_i32(buf)?; }
105 if version >= 2 { out.is_renew_ack = get_bool(buf)?; }
106 if version >= 0 { out.topics = { let n = crate::primitives::array::get_array_len(buf, flex)?; let mut v = Vec::with_capacity(n); for _ in 0..n { v.push(AcknowledgeTopic::decode_borrow(buf, version)?); } v }; }
107 if flex {
108 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| {
109 Ok(false)
110 })?;
111 }
112 Ok(out)
113 }
114}
115
116#[derive(Debug, Clone, PartialEq, Eq)]
117pub struct AcknowledgeTopic {
118 pub topic_id: crate::primitives::uuid::Uuid,
119 pub partitions: Vec<AcknowledgePartition>,
120 pub unknown_tagged_fields: UnknownTaggedFields,
121}
122
123impl Default for AcknowledgeTopic {
124 fn default() -> Self {
125 Self {
126 topic_id: Default::default(),
127 partitions: Vec::new(),
128 unknown_tagged_fields: Default::default(),
129 }
130 }
131}
132
133impl AcknowledgeTopic {
134 pub fn to_owned(&self) -> crate::owned::share_acknowledge_request::AcknowledgeTopic {
135 crate::owned::share_acknowledge_request::AcknowledgeTopic {
136 topic_id: (self.topic_id),
137 partitions: (self.partitions).iter().map(|it| it.to_owned()).collect(),
138 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
139 }
140 }
141}
142
143impl Encode for AcknowledgeTopic {
144 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
145 let flex = version >= 0;
146 if version >= 0 { crate::primitives::uuid::put_uuid(buf, self.topic_id) }
147 if version >= 0 { { crate::primitives::array::put_array_len(buf, (self.partitions).len(), flex); for it in &self.partitions { it.encode(buf, version)?; } } }
148 if flex {
149 let tagged = WriteTaggedFields::new();
150 tagged.write(buf, &self.unknown_tagged_fields);
151 }
152 Ok(())
153 }
154 fn encoded_len(&self, version: i16) -> usize {
155 let flex = version >= 0;
156 let mut n: usize = 0;
157 if version >= 0 { n += 16; }
158 if version >= 0 { n += { let prefix = crate::primitives::array::array_len_prefix_len((self.partitions).len(), flex); let body: usize = (self.partitions).iter().map(|it| it.encoded_len(version)).sum(); prefix + body }; }
159 if flex {
160 let known_pairs: Vec<(u32, usize)> = Vec::new();
161 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
162 }
163 n
164 }
165}
166
167impl<'de> DecodeBorrow<'de> for AcknowledgeTopic {
168 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
169 let flex = version >= 0;
170 let mut out = Self::default();
171 if version >= 0 { out.topic_id = crate::primitives::uuid::get_uuid(buf)?; }
172 if version >= 0 { out.partitions = { let n = crate::primitives::array::get_array_len(buf, flex)?; let mut v = Vec::with_capacity(n); for _ in 0..n { v.push(AcknowledgePartition::decode_borrow(buf, version)?); } v }; }
173 if flex {
174 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| {
175 Ok(false)
176 })?;
177 }
178 Ok(out)
179 }
180}
181
182#[derive(Debug, Clone, PartialEq, Eq)]
183pub struct AcknowledgePartition {
184 pub partition_index: i32,
185 pub acknowledgement_batches: Vec<AcknowledgementBatch>,
186 pub unknown_tagged_fields: UnknownTaggedFields,
187}
188
189impl Default for AcknowledgePartition {
190 fn default() -> Self {
191 Self {
192 partition_index: 0i32,
193 acknowledgement_batches: Vec::new(),
194 unknown_tagged_fields: Default::default(),
195 }
196 }
197}
198
199impl AcknowledgePartition {
200 pub fn to_owned(&self) -> crate::owned::share_acknowledge_request::AcknowledgePartition {
201 crate::owned::share_acknowledge_request::AcknowledgePartition {
202 partition_index: (self.partition_index),
203 acknowledgement_batches: (self.acknowledgement_batches).iter().map(|it| it.to_owned()).collect(),
204 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
205 }
206 }
207}
208
209impl Encode for AcknowledgePartition {
210 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
211 let flex = version >= 0;
212 if version >= 0 { put_i32(buf, self.partition_index) }
213 if version >= 0 { { crate::primitives::array::put_array_len(buf, (self.acknowledgement_batches).len(), flex); for it in &self.acknowledgement_batches { it.encode(buf, version)?; } } }
214 if flex {
215 let tagged = WriteTaggedFields::new();
216 tagged.write(buf, &self.unknown_tagged_fields);
217 }
218 Ok(())
219 }
220 fn encoded_len(&self, version: i16) -> usize {
221 let flex = version >= 0;
222 let mut n: usize = 0;
223 if version >= 0 { n += 4; }
224 if version >= 0 { n += { let prefix = crate::primitives::array::array_len_prefix_len((self.acknowledgement_batches).len(), flex); let body: usize = (self.acknowledgement_batches).iter().map(|it| it.encoded_len(version)).sum(); prefix + body }; }
225 if flex {
226 let known_pairs: Vec<(u32, usize)> = Vec::new();
227 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
228 }
229 n
230 }
231}
232
233impl<'de> DecodeBorrow<'de> for AcknowledgePartition {
234 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
235 let flex = version >= 0;
236 let mut out = Self::default();
237 if version >= 0 { out.partition_index = get_i32(buf)?; }
238 if version >= 0 { out.acknowledgement_batches = { let n = crate::primitives::array::get_array_len(buf, flex)?; let mut v = Vec::with_capacity(n); for _ in 0..n { v.push(AcknowledgementBatch::decode_borrow(buf, version)?); } v }; }
239 if flex {
240 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| {
241 Ok(false)
242 })?;
243 }
244 Ok(out)
245 }
246}
247
248#[derive(Debug, Clone, PartialEq, Eq)]
249pub struct AcknowledgementBatch {
250 pub first_offset: i64,
251 pub last_offset: i64,
252 pub acknowledge_types: Vec<i8>,
253 pub unknown_tagged_fields: UnknownTaggedFields,
254}
255
256impl Default for AcknowledgementBatch {
257 fn default() -> Self {
258 Self {
259 first_offset: 0i64,
260 last_offset: 0i64,
261 acknowledge_types: Vec::new(),
262 unknown_tagged_fields: Default::default(),
263 }
264 }
265}
266
267impl AcknowledgementBatch {
268 pub fn to_owned(&self) -> crate::owned::share_acknowledge_request::AcknowledgementBatch {
269 crate::owned::share_acknowledge_request::AcknowledgementBatch {
270 first_offset: (self.first_offset),
271 last_offset: (self.last_offset),
272 acknowledge_types: (self.acknowledge_types).clone(),
273 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
274 }
275 }
276}
277
278impl Encode for AcknowledgementBatch {
279 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
280 let flex = version >= 0;
281 if version >= 0 { put_i64(buf, self.first_offset) }
282 if version >= 0 { put_i64(buf, self.last_offset) }
283 if version >= 0 { { crate::primitives::array::put_array_len(buf, (self.acknowledge_types).len(), flex); for it in &self.acknowledge_types { put_i8(buf, *it); } } }
284 if flex {
285 let tagged = WriteTaggedFields::new();
286 tagged.write(buf, &self.unknown_tagged_fields);
287 }
288 Ok(())
289 }
290 fn encoded_len(&self, version: i16) -> usize {
291 let flex = version >= 0;
292 let mut n: usize = 0;
293 if version >= 0 { n += 8; }
294 if version >= 0 { n += 8; }
295 if version >= 0 { n += { let prefix = crate::primitives::array::array_len_prefix_len((self.acknowledge_types).len(), flex); let body: usize = (self.acknowledge_types).iter().map(|_| 1).sum(); prefix + body }; }
296 if flex {
297 let known_pairs: Vec<(u32, usize)> = Vec::new();
298 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
299 }
300 n
301 }
302}
303
304impl<'de> DecodeBorrow<'de> for AcknowledgementBatch {
305 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
306 let flex = version >= 0;
307 let mut out = Self::default();
308 if version >= 0 { out.first_offset = get_i64(buf)?; }
309 if version >= 0 { out.last_offset = get_i64(buf)?; }
310 if version >= 0 { out.acknowledge_types = { let n = crate::primitives::array::get_array_len(buf, flex)?; let mut v = Vec::with_capacity(n); for _ in 0..n { v.push(get_i8(buf)?); } v }; }
311 if flex {
312 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| {
313 Ok(false)
314 })?;
315 }
316 Ok(out)
317 }
318}