1use bytes::BufMut;
4
5use crate::primitives::fixed::{get_bool, get_i16, get_i32, get_i8, put_bool, put_i16, put_i32};
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 = 75;
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 DescribeTopicPartitionsResponse<'a> {
28 pub throttle_time_ms: i32,
29 pub topics: Vec<DescribeTopicPartitionsResponseTopic<'a>>,
30 pub next_cursor: Option<Cursor<'a>>,
31 pub unknown_tagged_fields: UnknownTaggedFields,
32}
33
34impl<'a> Default for DescribeTopicPartitionsResponse<'a> {
35 fn default() -> Self {
36 Self {
37 throttle_time_ms: 0i32,
38 topics: Vec::new(),
39 next_cursor: None,
40 unknown_tagged_fields: Default::default(),
41 }
42 }
43}
44
45impl<'a> DescribeTopicPartitionsResponse<'a> {
46 pub fn to_owned(&self) -> crate::owned::describe_topic_partitions_response::DescribeTopicPartitionsResponse {
47 crate::owned::describe_topic_partitions_response::DescribeTopicPartitionsResponse {
48 throttle_time_ms: (self.throttle_time_ms),
49 topics: (self.topics).iter().map(|it| it.to_owned()).collect(),
50 next_cursor: (self.next_cursor).as_ref().map(|v| v.to_owned()),
51 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
52 }
53 }
54}
55
56impl<'a> Encode for DescribeTopicPartitionsResponse<'a> {
57 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
58 if !(MIN_VERSION..=MAX_VERSION).contains(&version) {
59 return Err(ProtocolError::UnsupportedVersion { api_key: API_KEY, version });
60 }
61 let flex = is_flexible(version);
62 if version >= 0 { put_i32(buf, self.throttle_time_ms) }
63 if version >= 0 { { crate::primitives::array::put_array_len(buf, (self.topics).len(), flex); for it in &self.topics { it.encode(buf, version)?; } } }
64 if version >= 0 { match &self.next_cursor { None => { buf.put_i8(-1); }, Some(v) => { buf.put_i8(1); v.encode(buf, version)?; } } }
65 if flex {
66 let tagged = WriteTaggedFields::new();
67 tagged.write(buf, &self.unknown_tagged_fields);
68 }
69 Ok(())
70 }
71 fn encoded_len(&self, version: i16) -> usize {
72 let flex = is_flexible(version);
73 let mut n: usize = 0;
74 if version >= 0 { n += 4; }
75 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 }; }
76 if version >= 0 { n += 1 + self.next_cursor.as_ref().map_or(0, |v| v.encoded_len(version)); }
77 if flex {
78 let known_pairs: Vec<(u32, usize)> = Vec::new();
79 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
80 }
81 n
82 }
83}
84
85impl<'de> DecodeBorrow<'de> for DescribeTopicPartitionsResponse<'de> {
86 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
87 if !(MIN_VERSION..=MAX_VERSION).contains(&version) {
88 return Err(ProtocolError::UnsupportedVersion { api_key: API_KEY, version });
89 }
90 let flex = is_flexible(version);
91 let mut out = Self::default();
92 if version >= 0 { out.throttle_time_ms = get_i32(buf)?; }
93 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(DescribeTopicPartitionsResponseTopic::decode_borrow(buf, version)?); } v }; }
94 if version >= 0 { out.next_cursor = if get_i8(buf)? < 0 { None } else { Some(Cursor::decode_borrow(buf, version)?) }; }
95 if flex {
96 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| {
97 Ok(false)
98 })?;
99 }
100 Ok(out)
101 }
102}
103
104#[derive(Debug, Clone, PartialEq, Eq)]
105pub struct DescribeTopicPartitionsResponseTopic<'a> {
106 pub error_code: i16,
107 pub name: Option<&'a str>,
108 pub topic_id: crate::primitives::uuid::Uuid,
109 pub is_internal: bool,
110 pub partitions: Vec<DescribeTopicPartitionsResponsePartition>,
111 pub topic_authorized_operations: i32,
112 pub unknown_tagged_fields: UnknownTaggedFields,
113}
114
115impl<'a> Default for DescribeTopicPartitionsResponseTopic<'a> {
116 fn default() -> Self {
117 Self {
118 error_code: 0i16,
119 name: None,
120 topic_id: Default::default(),
121 is_internal: false,
122 partitions: Vec::new(),
123 topic_authorized_operations: -2_147_483_648i32,
124 unknown_tagged_fields: Default::default(),
125 }
126 }
127}
128
129impl<'a> DescribeTopicPartitionsResponseTopic<'a> {
130 pub fn to_owned(&self) -> crate::owned::describe_topic_partitions_response::DescribeTopicPartitionsResponseTopic {
131 crate::owned::describe_topic_partitions_response::DescribeTopicPartitionsResponseTopic {
132 error_code: (self.error_code),
133 name: (self.name).map(|s| s.to_string()),
134 topic_id: (self.topic_id),
135 is_internal: (self.is_internal),
136 partitions: (self.partitions).iter().map(|it| it.to_owned()).collect(),
137 topic_authorized_operations: (self.topic_authorized_operations),
138 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
139 }
140 }
141}
142
143impl<'a> Encode for DescribeTopicPartitionsResponseTopic<'a> {
144 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
145 let flex = version >= 0;
146 if version >= 0 { put_i16(buf, self.error_code) }
147 if version >= 0 { if flex { put_compact_nullable_string(buf, self.name) } else { put_nullable_string(buf, self.name) } }
148 if version >= 0 { crate::primitives::uuid::put_uuid(buf, self.topic_id) }
149 if version >= 0 { put_bool(buf, self.is_internal) }
150 if version >= 0 { { crate::primitives::array::put_array_len(buf, (self.partitions).len(), flex); for it in &self.partitions { it.encode(buf, version)?; } } }
151 if version >= 0 { put_i32(buf, self.topic_authorized_operations) }
152 if flex {
153 let tagged = WriteTaggedFields::new();
154 tagged.write(buf, &self.unknown_tagged_fields);
155 }
156 Ok(())
157 }
158 fn encoded_len(&self, version: i16) -> usize {
159 let flex = version >= 0;
160 let mut n: usize = 0;
161 if version >= 0 { n += 2; }
162 if version >= 0 { n += if flex { compact_nullable_string_len(self.name) } else { nullable_string_len(self.name) }; }
163 if version >= 0 { n += 16; }
164 if version >= 0 { n += 1; }
165 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 }; }
166 if version >= 0 { n += 4; }
167 if flex {
168 let known_pairs: Vec<(u32, usize)> = Vec::new();
169 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
170 }
171 n
172 }
173}
174
175impl<'de> DecodeBorrow<'de> for DescribeTopicPartitionsResponseTopic<'de> {
176 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
177 let flex = version >= 0;
178 let mut out = Self::default();
179 if version >= 0 { out.error_code = get_i16(buf)?; }
180 if version >= 0 { out.name = if flex { get_compact_nullable_string_borrowed(buf)? } else { get_nullable_string_borrowed(buf)? }; }
181 if version >= 0 { out.topic_id = crate::primitives::uuid::get_uuid(buf)?; }
182 if version >= 0 { out.is_internal = get_bool(buf)?; }
183 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(DescribeTopicPartitionsResponsePartition::decode_borrow(buf, version)?); } v }; }
184 if version >= 0 { out.topic_authorized_operations = get_i32(buf)?; }
185 if flex {
186 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| {
187 Ok(false)
188 })?;
189 }
190 Ok(out)
191 }
192}
193
194#[derive(Debug, Clone, PartialEq, Eq)]
195pub struct DescribeTopicPartitionsResponsePartition {
196 pub error_code: i16,
197 pub partition_index: i32,
198 pub leader_id: i32,
199 pub leader_epoch: i32,
200 pub replica_nodes: Vec<i32>,
201 pub isr_nodes: Vec<i32>,
202 pub eligible_leader_replicas: Option<Vec<i32>>,
203 pub last_known_elr: Option<Vec<i32>>,
204 pub offline_replicas: Vec<i32>,
205 pub unknown_tagged_fields: UnknownTaggedFields,
206}
207
208impl Default for DescribeTopicPartitionsResponsePartition {
209 fn default() -> Self {
210 Self {
211 error_code: 0i16,
212 partition_index: 0i32,
213 leader_id: 0i32,
214 leader_epoch: -1i32,
215 replica_nodes: Vec::new(),
216 isr_nodes: Vec::new(),
217 eligible_leader_replicas: None,
218 last_known_elr: None,
219 offline_replicas: Vec::new(),
220 unknown_tagged_fields: Default::default(),
221 }
222 }
223}
224
225impl DescribeTopicPartitionsResponsePartition {
226 pub fn to_owned(&self) -> crate::owned::describe_topic_partitions_response::DescribeTopicPartitionsResponsePartition {
227 crate::owned::describe_topic_partitions_response::DescribeTopicPartitionsResponsePartition {
228 error_code: (self.error_code),
229 partition_index: (self.partition_index),
230 leader_id: (self.leader_id),
231 leader_epoch: (self.leader_epoch),
232 replica_nodes: (self.replica_nodes).clone(),
233 isr_nodes: (self.isr_nodes).clone(),
234 eligible_leader_replicas: (self.eligible_leader_replicas).clone(),
235 last_known_elr: (self.last_known_elr).clone(),
236 offline_replicas: (self.offline_replicas).clone(),
237 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
238 }
239 }
240}
241
242impl Encode for DescribeTopicPartitionsResponsePartition {
243 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
244 let flex = version >= 0;
245 if version >= 0 { put_i16(buf, self.error_code) }
246 if version >= 0 { put_i32(buf, self.partition_index) }
247 if version >= 0 { put_i32(buf, self.leader_id) }
248 if version >= 0 { put_i32(buf, self.leader_epoch) }
249 if version >= 0 { { crate::primitives::array::put_array_len(buf, (self.replica_nodes).len(), flex); for it in &self.replica_nodes { put_i32(buf, *it); } } }
250 if version >= 0 { { crate::primitives::array::put_array_len(buf, (self.isr_nodes).len(), flex); for it in &self.isr_nodes { put_i32(buf, *it); } } }
251 if version >= 0 { { let len = (self.eligible_leader_replicas).as_ref().map(Vec::len); crate::primitives::array::put_nullable_array_len(buf, len, flex); if let Some(v) = &self.eligible_leader_replicas { for it in v { put_i32(buf, *it); } } } }
252 if version >= 0 { { let len = (self.last_known_elr).as_ref().map(Vec::len); crate::primitives::array::put_nullable_array_len(buf, len, flex); if let Some(v) = &self.last_known_elr { for it in v { put_i32(buf, *it); } } } }
253 if version >= 0 { { crate::primitives::array::put_array_len(buf, (self.offline_replicas).len(), flex); for it in &self.offline_replicas { put_i32(buf, *it); } } }
254 if flex {
255 let tagged = WriteTaggedFields::new();
256 tagged.write(buf, &self.unknown_tagged_fields);
257 }
258 Ok(())
259 }
260 fn encoded_len(&self, version: i16) -> usize {
261 let flex = version >= 0;
262 let mut n: usize = 0;
263 if version >= 0 { n += 2; }
264 if version >= 0 { n += 4; }
265 if version >= 0 { n += 4; }
266 if version >= 0 { n += 4; }
267 if version >= 0 { n += { let prefix = crate::primitives::array::array_len_prefix_len((self.replica_nodes).len(), flex); let body: usize = (self.replica_nodes).iter().map(|_| 4).sum(); prefix + body }; }
268 if version >= 0 { n += { let prefix = crate::primitives::array::array_len_prefix_len((self.isr_nodes).len(), flex); let body: usize = (self.isr_nodes).iter().map(|_| 4).sum(); prefix + body }; }
269 if version >= 0 { n += { let opt: Option<&Vec<_>> = (self.eligible_leader_replicas).as_ref(); let prefix = crate::primitives::array::nullable_array_len_prefix_len(opt.map(|v| v.len()), flex); let body: usize = opt.map_or(0, |v| v.iter().map(|_| 4).sum()); prefix + body }; }
270 if version >= 0 { n += { let opt: Option<&Vec<_>> = (self.last_known_elr).as_ref(); let prefix = crate::primitives::array::nullable_array_len_prefix_len(opt.map(|v| v.len()), flex); let body: usize = opt.map_or(0, |v| v.iter().map(|_| 4).sum()); prefix + body }; }
271 if version >= 0 { n += { let prefix = crate::primitives::array::array_len_prefix_len((self.offline_replicas).len(), flex); let body: usize = (self.offline_replicas).iter().map(|_| 4).sum(); prefix + body }; }
272 if flex {
273 let known_pairs: Vec<(u32, usize)> = Vec::new();
274 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
275 }
276 n
277 }
278}
279
280impl<'de> DecodeBorrow<'de> for DescribeTopicPartitionsResponsePartition {
281 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
282 let flex = version >= 0;
283 let mut out = Self::default();
284 if version >= 0 { out.error_code = get_i16(buf)?; }
285 if version >= 0 { out.partition_index = get_i32(buf)?; }
286 if version >= 0 { out.leader_id = get_i32(buf)?; }
287 if version >= 0 { out.leader_epoch = get_i32(buf)?; }
288 if version >= 0 { out.replica_nodes = { let n = crate::primitives::array::get_array_len(buf, flex)?; let mut v = Vec::with_capacity(n); for _ in 0..n { v.push(get_i32(buf)?); } v }; }
289 if version >= 0 { out.isr_nodes = { let n = crate::primitives::array::get_array_len(buf, flex)?; let mut v = Vec::with_capacity(n); for _ in 0..n { v.push(get_i32(buf)?); } v }; }
290 if version >= 0 { out.eligible_leader_replicas = { let opt = crate::primitives::array::get_nullable_array_len(buf, flex)?; match opt { None => None, Some(n) => { let mut v = Vec::with_capacity(n); for _ in 0..n { v.push(get_i32(buf)?); } Some(v) } } }; }
291 if version >= 0 { out.last_known_elr = { let opt = crate::primitives::array::get_nullable_array_len(buf, flex)?; match opt { None => None, Some(n) => { let mut v = Vec::with_capacity(n); for _ in 0..n { v.push(get_i32(buf)?); } Some(v) } } }; }
292 if version >= 0 { out.offline_replicas = { let n = crate::primitives::array::get_array_len(buf, flex)?; let mut v = Vec::with_capacity(n); for _ in 0..n { v.push(get_i32(buf)?); } v }; }
293 if flex {
294 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| {
295 Ok(false)
296 })?;
297 }
298 Ok(out)
299 }
300}
301
302#[derive(Debug, Clone, PartialEq, Eq)]
303pub struct Cursor<'a> {
304 pub topic_name: &'a str,
305 pub partition_index: i32,
306 pub unknown_tagged_fields: UnknownTaggedFields,
307}
308
309impl<'a> Default for Cursor<'a> {
310 fn default() -> Self {
311 Self {
312 topic_name: "",
313 partition_index: 0i32,
314 unknown_tagged_fields: Default::default(),
315 }
316 }
317}
318
319impl<'a> Cursor<'a> {
320 pub fn to_owned(&self) -> crate::owned::describe_topic_partitions_response::Cursor {
321 crate::owned::describe_topic_partitions_response::Cursor {
322 topic_name: (self.topic_name).to_string(),
323 partition_index: (self.partition_index),
324 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
325 }
326 }
327}
328
329impl<'a> Encode for Cursor<'a> {
330 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
331 let flex = version >= 0;
332 if version >= 0 { if flex { put_compact_string(buf, self.topic_name) } else { put_string(buf, self.topic_name) } }
333 if version >= 0 { put_i32(buf, self.partition_index) }
334 if flex {
335 let tagged = WriteTaggedFields::new();
336 tagged.write(buf, &self.unknown_tagged_fields);
337 }
338 Ok(())
339 }
340 fn encoded_len(&self, version: i16) -> usize {
341 let flex = version >= 0;
342 let mut n: usize = 0;
343 if version >= 0 { n += if flex { compact_string_len(self.topic_name) } else { string_len(self.topic_name) }; }
344 if version >= 0 { n += 4; }
345 if flex {
346 let known_pairs: Vec<(u32, usize)> = Vec::new();
347 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
348 }
349 n
350 }
351}
352
353impl<'de> DecodeBorrow<'de> for Cursor<'de> {
354 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
355 let flex = version >= 0;
356 let mut out = Self::default();
357 if version >= 0 { out.topic_name = if flex { get_compact_string_borrowed(buf)? } else { get_string_borrowed(buf)? }; }
358 if version >= 0 { out.partition_index = get_i32(buf)?; }
359 if flex {
360 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| {
361 Ok(false)
362 })?;
363 }
364 Ok(out)
365 }
366}