1use bytes::BufMut;
4
5use crate::primitives::fixed::{get_bool, get_i16, get_i32, get_i64, get_u16, put_bool, put_i16, put_i32, put_i64, put_u16};
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 = 62;
19pub const MIN_VERSION: i16 = 0;
20pub const MAX_VERSION: i16 = 4;
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 BrokerRegistrationRequest<'a> {
28 pub broker_id: i32,
29 pub cluster_id: &'a str,
30 pub incarnation_id: crate::primitives::uuid::Uuid,
31 pub listeners: Vec<Listener<'a>>,
32 pub features: Vec<Feature<'a>>,
33 pub rack: Option<&'a str>,
34 pub is_migrating_zk_broker: bool,
35 pub log_dirs: Vec<crate::primitives::uuid::Uuid>,
36 pub previous_broker_epoch: i64,
37 pub unknown_tagged_fields: UnknownTaggedFields,
38}
39
40impl<'a> Default for BrokerRegistrationRequest<'a> {
41 fn default() -> Self {
42 Self {
43 broker_id: 0i32,
44 cluster_id: "",
45 incarnation_id: Default::default(),
46 listeners: Vec::new(),
47 features: Vec::new(),
48 rack: None,
49 is_migrating_zk_broker: false,
50 log_dirs: Vec::new(),
51 previous_broker_epoch: -1i64,
52 unknown_tagged_fields: Default::default(),
53 }
54 }
55}
56
57impl<'a> BrokerRegistrationRequest<'a> {
58 pub fn to_owned(&self) -> crate::owned::broker_registration_request::BrokerRegistrationRequest {
59 crate::owned::broker_registration_request::BrokerRegistrationRequest {
60 broker_id: (self.broker_id),
61 cluster_id: (self.cluster_id).to_string(),
62 incarnation_id: (self.incarnation_id),
63 listeners: (self.listeners).iter().map(|it| it.to_owned()).collect(),
64 features: (self.features).iter().map(|it| it.to_owned()).collect(),
65 rack: (self.rack).map(|s| s.to_string()),
66 is_migrating_zk_broker: (self.is_migrating_zk_broker),
67 log_dirs: (self.log_dirs).clone(),
68 previous_broker_epoch: (self.previous_broker_epoch),
69 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
70 }
71 }
72}
73
74impl<'a> Encode for BrokerRegistrationRequest<'a> {
75 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
76 if !(MIN_VERSION..=MAX_VERSION).contains(&version) {
77 return Err(ProtocolError::UnsupportedVersion { api_key: API_KEY, version });
78 }
79 let flex = is_flexible(version);
80 if version >= 0 { put_i32(buf, self.broker_id) }
81 if version >= 0 { if flex { put_compact_string(buf, self.cluster_id) } else { put_string(buf, self.cluster_id) } }
82 if version >= 0 { crate::primitives::uuid::put_uuid(buf, self.incarnation_id) }
83 if version >= 0 { { crate::primitives::array::put_array_len(buf, (self.listeners).len(), flex); for it in &self.listeners { it.encode(buf, version)?; } } }
84 if version >= 0 { { crate::primitives::array::put_array_len(buf, (self.features).len(), flex); for it in &self.features { it.encode(buf, version)?; } } }
85 if version >= 0 { if flex { put_compact_nullable_string(buf, self.rack) } else { put_nullable_string(buf, self.rack) } }
86 if version >= 1 { put_bool(buf, self.is_migrating_zk_broker) }
87 if version >= 2 { { crate::primitives::array::put_array_len(buf, (self.log_dirs).len(), flex); for it in &self.log_dirs { crate::primitives::uuid::put_uuid(buf, *it); } } }
88 if version >= 3 { put_i64(buf, self.previous_broker_epoch) }
89 if flex {
90 let tagged = WriteTaggedFields::new();
91 tagged.write(buf, &self.unknown_tagged_fields);
92 }
93 Ok(())
94 }
95 fn encoded_len(&self, version: i16) -> usize {
96 let flex = is_flexible(version);
97 let mut n: usize = 0;
98 if version >= 0 { n += 4; }
99 if version >= 0 { n += if flex { compact_string_len(self.cluster_id) } else { string_len(self.cluster_id) }; }
100 if version >= 0 { n += 16; }
101 if version >= 0 { n += { let prefix = crate::primitives::array::array_len_prefix_len((self.listeners).len(), flex); let body: usize = (self.listeners).iter().map(|it| it.encoded_len(version)).sum(); prefix + body }; }
102 if version >= 0 { n += { let prefix = crate::primitives::array::array_len_prefix_len((self.features).len(), flex); let body: usize = (self.features).iter().map(|it| it.encoded_len(version)).sum(); prefix + body }; }
103 if version >= 0 { n += if flex { compact_nullable_string_len(self.rack) } else { nullable_string_len(self.rack) }; }
104 if version >= 1 { n += 1; }
105 if version >= 2 { n += { let prefix = crate::primitives::array::array_len_prefix_len((self.log_dirs).len(), flex); let body: usize = (self.log_dirs).iter().map(|_| 16).sum(); prefix + body }; }
106 if version >= 3 { n += 8; }
107 if flex {
108 let known_pairs: Vec<(u32, usize)> = Vec::new();
109 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
110 }
111 n
112 }
113}
114
115impl<'de> DecodeBorrow<'de> for BrokerRegistrationRequest<'de> {
116 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
117 if !(MIN_VERSION..=MAX_VERSION).contains(&version) {
118 return Err(ProtocolError::UnsupportedVersion { api_key: API_KEY, version });
119 }
120 let flex = is_flexible(version);
121 let mut out = Self::default();
122 if version >= 0 { out.broker_id = get_i32(buf)?; }
123 if version >= 0 { out.cluster_id = if flex { get_compact_string_borrowed(buf)? } else { get_string_borrowed(buf)? }; }
124 if version >= 0 { out.incarnation_id = crate::primitives::uuid::get_uuid(buf)?; }
125 if version >= 0 { out.listeners = { let n = crate::primitives::array::get_array_len(buf, flex)?; let mut v = Vec::with_capacity(n); for _ in 0..n { v.push(Listener::decode_borrow(buf, version)?); } v }; }
126 if version >= 0 { out.features = { let n = crate::primitives::array::get_array_len(buf, flex)?; let mut v = Vec::with_capacity(n); for _ in 0..n { v.push(Feature::decode_borrow(buf, version)?); } v }; }
127 if version >= 0 { out.rack = if flex { get_compact_nullable_string_borrowed(buf)? } else { get_nullable_string_borrowed(buf)? }; }
128 if version >= 1 { out.is_migrating_zk_broker = get_bool(buf)?; }
129 if version >= 2 { out.log_dirs = { let n = crate::primitives::array::get_array_len(buf, flex)?; let mut v = Vec::with_capacity(n); for _ in 0..n { v.push(crate::primitives::uuid::get_uuid(buf)?); } v }; }
130 if version >= 3 { out.previous_broker_epoch = get_i64(buf)?; }
131 if flex {
132 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| {
133 Ok(false)
134 })?;
135 }
136 Ok(out)
137 }
138}
139
140#[derive(Debug, Clone, PartialEq, Eq)]
141pub struct Listener<'a> {
142 pub name: &'a str,
143 pub host: &'a str,
144 pub port: u16,
145 pub security_protocol: i16,
146 pub unknown_tagged_fields: UnknownTaggedFields,
147}
148
149impl<'a> Default for Listener<'a> {
150 fn default() -> Self {
151 Self {
152 name: "",
153 host: "",
154 port: 0u16,
155 security_protocol: 0i16,
156 unknown_tagged_fields: Default::default(),
157 }
158 }
159}
160
161impl<'a> Listener<'a> {
162 pub fn to_owned(&self) -> crate::owned::broker_registration_request::Listener {
163 crate::owned::broker_registration_request::Listener {
164 name: (self.name).to_string(),
165 host: (self.host).to_string(),
166 port: (self.port),
167 security_protocol: (self.security_protocol),
168 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
169 }
170 }
171}
172
173impl<'a> Encode for Listener<'a> {
174 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
175 let flex = version >= 0;
176 if version >= 0 { if flex { put_compact_string(buf, self.name) } else { put_string(buf, self.name) } }
177 if version >= 0 { if flex { put_compact_string(buf, self.host) } else { put_string(buf, self.host) } }
178 if version >= 0 { put_u16(buf, self.port) }
179 if version >= 0 { put_i16(buf, self.security_protocol) }
180 if flex {
181 let tagged = WriteTaggedFields::new();
182 tagged.write(buf, &self.unknown_tagged_fields);
183 }
184 Ok(())
185 }
186 fn encoded_len(&self, version: i16) -> usize {
187 let flex = version >= 0;
188 let mut n: usize = 0;
189 if version >= 0 { n += if flex { compact_string_len(self.name) } else { string_len(self.name) }; }
190 if version >= 0 { n += if flex { compact_string_len(self.host) } else { string_len(self.host) }; }
191 if version >= 0 { n += 2; }
192 if version >= 0 { n += 2; }
193 if flex {
194 let known_pairs: Vec<(u32, usize)> = Vec::new();
195 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
196 }
197 n
198 }
199}
200
201impl<'de> DecodeBorrow<'de> for Listener<'de> {
202 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
203 let flex = version >= 0;
204 let mut out = Self::default();
205 if version >= 0 { out.name = if flex { get_compact_string_borrowed(buf)? } else { get_string_borrowed(buf)? }; }
206 if version >= 0 { out.host = if flex { get_compact_string_borrowed(buf)? } else { get_string_borrowed(buf)? }; }
207 if version >= 0 { out.port = get_u16(buf)?; }
208 if version >= 0 { out.security_protocol = get_i16(buf)?; }
209 if flex {
210 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| {
211 Ok(false)
212 })?;
213 }
214 Ok(out)
215 }
216}
217
218#[derive(Debug, Clone, PartialEq, Eq)]
219pub struct Feature<'a> {
220 pub name: &'a str,
221 pub min_supported_version: i16,
222 pub max_supported_version: i16,
223 pub unknown_tagged_fields: UnknownTaggedFields,
224}
225
226impl<'a> Default for Feature<'a> {
227 fn default() -> Self {
228 Self {
229 name: "",
230 min_supported_version: 0i16,
231 max_supported_version: 0i16,
232 unknown_tagged_fields: Default::default(),
233 }
234 }
235}
236
237impl<'a> Feature<'a> {
238 pub fn to_owned(&self) -> crate::owned::broker_registration_request::Feature {
239 crate::owned::broker_registration_request::Feature {
240 name: (self.name).to_string(),
241 min_supported_version: (self.min_supported_version),
242 max_supported_version: (self.max_supported_version),
243 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
244 }
245 }
246}
247
248impl<'a> Encode for Feature<'a> {
249 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
250 let flex = version >= 0;
251 if version >= 0 { if flex { put_compact_string(buf, self.name) } else { put_string(buf, self.name) } }
252 if version >= 0 { put_i16(buf, self.min_supported_version) }
253 if version >= 0 { put_i16(buf, self.max_supported_version) }
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 += if flex { compact_string_len(self.name) } else { string_len(self.name) }; }
264 if version >= 0 { n += 2; }
265 if version >= 0 { n += 2; }
266 if flex {
267 let known_pairs: Vec<(u32, usize)> = Vec::new();
268 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
269 }
270 n
271 }
272}
273
274impl<'de> DecodeBorrow<'de> for Feature<'de> {
275 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
276 let flex = version >= 0;
277 let mut out = Self::default();
278 if version >= 0 { out.name = if flex { get_compact_string_borrowed(buf)? } else { get_string_borrowed(buf)? }; }
279 if version >= 0 { out.min_supported_version = get_i16(buf)?; }
280 if version >= 0 { out.max_supported_version = get_i16(buf)?; }
281 if flex {
282 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| {
283 Ok(false)
284 })?;
285 }
286 Ok(out)
287 }
288}