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