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