1#![allow(unused)]
6
7use std::borrow::Borrow;
8use std::collections::BTreeMap;
9
10use anyhow::{bail, Result};
11use bytes::Bytes;
12use uuid::Uuid;
13
14use crate::protocol::{
15 buf::{ByteBuf, ByteBufMut},
16 compute_unknown_tagged_fields_size, types, write_unknown_tagged_fields, Decodable, Decoder,
17 Encodable, Encoder, HeaderVersion, Message, StrBytes, VersionRange,
18};
19
20#[non_exhaustive]
22#[derive(Debug, Clone, PartialEq)]
23pub struct ListTransactionsResponse {
24 pub throttle_time_ms: i32,
28
29 pub error_code: i16,
33
34 pub unknown_state_filters: Vec<StrBytes>,
38
39 pub transaction_states: Vec<TransactionState>,
43
44 pub unknown_tagged_fields: BTreeMap<i32, Bytes>,
46}
47
48impl ListTransactionsResponse {
49 pub fn with_throttle_time_ms(mut self, value: i32) -> Self {
55 self.throttle_time_ms = value;
56 self
57 }
58 pub fn with_error_code(mut self, value: i16) -> Self {
64 self.error_code = value;
65 self
66 }
67 pub fn with_unknown_state_filters(mut self, value: Vec<StrBytes>) -> Self {
73 self.unknown_state_filters = value;
74 self
75 }
76 pub fn with_transaction_states(mut self, value: Vec<TransactionState>) -> Self {
82 self.transaction_states = value;
83 self
84 }
85 pub fn with_unknown_tagged_fields(mut self, value: BTreeMap<i32, Bytes>) -> Self {
87 self.unknown_tagged_fields = value;
88 self
89 }
90 pub fn with_unknown_tagged_field(mut self, key: i32, value: Bytes) -> Self {
92 self.unknown_tagged_fields.insert(key, value);
93 self
94 }
95}
96
97#[cfg(feature = "broker")]
98impl Encodable for ListTransactionsResponse {
99 fn encode<B: ByteBufMut>(&self, buf: &mut B, version: i16) -> Result<()> {
100 if version < 0 || version > 1 {
101 bail!("specified version not supported by this message type");
102 }
103 types::Int32.encode(buf, &self.throttle_time_ms)?;
104 types::Int16.encode(buf, &self.error_code)?;
105 types::CompactArray(types::CompactString).encode(buf, &self.unknown_state_filters)?;
106 types::CompactArray(types::Struct { version }).encode(buf, &self.transaction_states)?;
107 let num_tagged_fields = self.unknown_tagged_fields.len();
108 if num_tagged_fields > std::u32::MAX as usize {
109 bail!(
110 "Too many tagged fields to encode ({} fields)",
111 num_tagged_fields
112 );
113 }
114 types::UnsignedVarInt.encode(buf, num_tagged_fields as u32)?;
115
116 write_unknown_tagged_fields(buf, 0.., &self.unknown_tagged_fields)?;
117 Ok(())
118 }
119 fn compute_size(&self, version: i16) -> Result<usize> {
120 let mut total_size = 0;
121 total_size += types::Int32.compute_size(&self.throttle_time_ms)?;
122 total_size += types::Int16.compute_size(&self.error_code)?;
123 total_size +=
124 types::CompactArray(types::CompactString).compute_size(&self.unknown_state_filters)?;
125 total_size += types::CompactArray(types::Struct { version })
126 .compute_size(&self.transaction_states)?;
127 let num_tagged_fields = self.unknown_tagged_fields.len();
128 if num_tagged_fields > std::u32::MAX as usize {
129 bail!(
130 "Too many tagged fields to encode ({} fields)",
131 num_tagged_fields
132 );
133 }
134 total_size += types::UnsignedVarInt.compute_size(num_tagged_fields as u32)?;
135
136 total_size += compute_unknown_tagged_fields_size(&self.unknown_tagged_fields)?;
137 Ok(total_size)
138 }
139}
140
141#[cfg(feature = "client")]
142impl Decodable for ListTransactionsResponse {
143 fn decode<B: ByteBuf>(buf: &mut B, version: i16) -> Result<Self> {
144 if version < 0 || version > 1 {
145 bail!("specified version not supported by this message type");
146 }
147 let throttle_time_ms = types::Int32.decode(buf)?;
148 let error_code = types::Int16.decode(buf)?;
149 let unknown_state_filters = types::CompactArray(types::CompactString).decode(buf)?;
150 let transaction_states = types::CompactArray(types::Struct { version }).decode(buf)?;
151 let mut unknown_tagged_fields = BTreeMap::new();
152 let num_tagged_fields = types::UnsignedVarInt.decode(buf)?;
153 for _ in 0..num_tagged_fields {
154 let tag: u32 = types::UnsignedVarInt.decode(buf)?;
155 let size: u32 = types::UnsignedVarInt.decode(buf)?;
156 let unknown_value = buf.try_get_bytes(size as usize)?;
157 unknown_tagged_fields.insert(tag as i32, unknown_value);
158 }
159 Ok(Self {
160 throttle_time_ms,
161 error_code,
162 unknown_state_filters,
163 transaction_states,
164 unknown_tagged_fields,
165 })
166 }
167}
168
169impl Default for ListTransactionsResponse {
170 fn default() -> Self {
171 Self {
172 throttle_time_ms: 0,
173 error_code: 0,
174 unknown_state_filters: Default::default(),
175 transaction_states: Default::default(),
176 unknown_tagged_fields: BTreeMap::new(),
177 }
178 }
179}
180
181impl Message for ListTransactionsResponse {
182 const VERSIONS: VersionRange = VersionRange { min: 0, max: 1 };
183 const DEPRECATED_VERSIONS: Option<VersionRange> = None;
184}
185
186#[non_exhaustive]
188#[derive(Debug, Clone, PartialEq)]
189pub struct TransactionState {
190 pub transactional_id: super::TransactionalId,
194
195 pub producer_id: super::ProducerId,
199
200 pub transaction_state: StrBytes,
204
205 pub unknown_tagged_fields: BTreeMap<i32, Bytes>,
207}
208
209impl TransactionState {
210 pub fn with_transactional_id(mut self, value: super::TransactionalId) -> Self {
216 self.transactional_id = value;
217 self
218 }
219 pub fn with_producer_id(mut self, value: super::ProducerId) -> Self {
225 self.producer_id = value;
226 self
227 }
228 pub fn with_transaction_state(mut self, value: StrBytes) -> Self {
234 self.transaction_state = value;
235 self
236 }
237 pub fn with_unknown_tagged_fields(mut self, value: BTreeMap<i32, Bytes>) -> Self {
239 self.unknown_tagged_fields = value;
240 self
241 }
242 pub fn with_unknown_tagged_field(mut self, key: i32, value: Bytes) -> Self {
244 self.unknown_tagged_fields.insert(key, value);
245 self
246 }
247}
248
249#[cfg(feature = "broker")]
250impl Encodable for TransactionState {
251 fn encode<B: ByteBufMut>(&self, buf: &mut B, version: i16) -> Result<()> {
252 if version < 0 || version > 1 {
253 bail!("specified version not supported by this message type");
254 }
255 types::CompactString.encode(buf, &self.transactional_id)?;
256 types::Int64.encode(buf, &self.producer_id)?;
257 types::CompactString.encode(buf, &self.transaction_state)?;
258 let num_tagged_fields = self.unknown_tagged_fields.len();
259 if num_tagged_fields > std::u32::MAX as usize {
260 bail!(
261 "Too many tagged fields to encode ({} fields)",
262 num_tagged_fields
263 );
264 }
265 types::UnsignedVarInt.encode(buf, num_tagged_fields as u32)?;
266
267 write_unknown_tagged_fields(buf, 0.., &self.unknown_tagged_fields)?;
268 Ok(())
269 }
270 fn compute_size(&self, version: i16) -> Result<usize> {
271 let mut total_size = 0;
272 total_size += types::CompactString.compute_size(&self.transactional_id)?;
273 total_size += types::Int64.compute_size(&self.producer_id)?;
274 total_size += types::CompactString.compute_size(&self.transaction_state)?;
275 let num_tagged_fields = self.unknown_tagged_fields.len();
276 if num_tagged_fields > std::u32::MAX as usize {
277 bail!(
278 "Too many tagged fields to encode ({} fields)",
279 num_tagged_fields
280 );
281 }
282 total_size += types::UnsignedVarInt.compute_size(num_tagged_fields as u32)?;
283
284 total_size += compute_unknown_tagged_fields_size(&self.unknown_tagged_fields)?;
285 Ok(total_size)
286 }
287}
288
289#[cfg(feature = "client")]
290impl Decodable for TransactionState {
291 fn decode<B: ByteBuf>(buf: &mut B, version: i16) -> Result<Self> {
292 if version < 0 || version > 1 {
293 bail!("specified version not supported by this message type");
294 }
295 let transactional_id = types::CompactString.decode(buf)?;
296 let producer_id = types::Int64.decode(buf)?;
297 let transaction_state = types::CompactString.decode(buf)?;
298 let mut unknown_tagged_fields = BTreeMap::new();
299 let num_tagged_fields = types::UnsignedVarInt.decode(buf)?;
300 for _ in 0..num_tagged_fields {
301 let tag: u32 = types::UnsignedVarInt.decode(buf)?;
302 let size: u32 = types::UnsignedVarInt.decode(buf)?;
303 let unknown_value = buf.try_get_bytes(size as usize)?;
304 unknown_tagged_fields.insert(tag as i32, unknown_value);
305 }
306 Ok(Self {
307 transactional_id,
308 producer_id,
309 transaction_state,
310 unknown_tagged_fields,
311 })
312 }
313}
314
315impl Default for TransactionState {
316 fn default() -> Self {
317 Self {
318 transactional_id: Default::default(),
319 producer_id: (0).into(),
320 transaction_state: Default::default(),
321 unknown_tagged_fields: BTreeMap::new(),
322 }
323 }
324}
325
326impl Message for TransactionState {
327 const VERSIONS: VersionRange = VersionRange { min: 0, max: 1 };
328 const DEPRECATED_VERSIONS: Option<VersionRange> = None;
329}
330
331impl HeaderVersion for ListTransactionsResponse {
332 fn header_version(version: i16) -> i16 {
333 1
334 }
335}