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