1use bytes::{BufMut, Bytes};
3
4use crate::{
5 DecodeBorrow, Encode, ProtocolError, UnknownTaggedFields,
6 primitives::{
7 fixed::{get_bool, get_i16, get_i32, put_bool, put_i16, put_i32},
8 string_bytes::{
9 compact_nullable_string_len, compact_string_len, nullable_string_len, put_bytes,
10 put_compact_bytes, put_compact_nullable_string, put_compact_string,
11 put_nullable_string, put_string, string_len,
12 },
13 string_bytes_borrowed::{
14 get_bytes_borrowed, get_compact_bytes_borrowed, get_compact_nullable_string_borrowed,
15 get_compact_string_borrowed, get_nullable_string_borrowed, get_string_borrowed,
16 },
17 },
18 tagged_fields::{WriteTaggedFields, read_tagged_fields, tagged_fields_len},
19};
20pub const API_KEY: i16 = 11;
21pub const MIN_VERSION: i16 = 0;
22pub const MAX_VERSION: i16 = 9;
23pub const FLEXIBLE_MIN: i16 = 6;
24#[inline]
25#[must_use]
26pub fn is_flexible(version: i16) -> bool {
27 version >= FLEXIBLE_MIN
28}
29#[derive(Debug, Clone, PartialEq, Eq)]
30pub struct JoinGroupResponse<'a> {
31 pub throttle_time_ms: i32,
32 pub error_code: i16,
33 pub generation_id: i32,
34 pub protocol_type: Option<&'a str>,
35 pub protocol_name: Option<&'a str>,
36 pub leader: &'a str,
37 pub skip_assignment: bool,
38 pub member_id: &'a str,
39 pub members: Vec<JoinGroupResponseMember<'a>>,
40 pub unknown_tagged_fields: UnknownTaggedFields,
41}
42impl Default for JoinGroupResponse<'_> {
43 fn default() -> Self {
44 Self {
45 throttle_time_ms: 0i32,
46 error_code: 0i16,
47 generation_id: -1i32,
48 protocol_type: None,
49 protocol_name: Some(""),
50 leader: "",
51 skip_assignment: false,
52 member_id: "",
53 members: Vec::new(),
54 unknown_tagged_fields: UnknownTaggedFields::default(),
55 }
56 }
57}
58impl<'a> JoinGroupResponse<'a> {
59 #[must_use]
63 pub fn to_owned(&self) -> crate::owned::join_group_response::JoinGroupResponse {
64 crate::owned::join_group_response::JoinGroupResponse {
65 throttle_time_ms: (self.throttle_time_ms),
66 error_code: (self.error_code),
67 generation_id: (self.generation_id),
68 protocol_type: (self.protocol_type).map(std::string::ToString::to_string),
69 protocol_name: (self.protocol_name).map(std::string::ToString::to_string),
70 leader: (self.leader).to_string(),
71 skip_assignment: (self.skip_assignment),
72 member_id: (self.member_id).to_string(),
73 members: (self.members)
74 .iter()
75 .map(JoinGroupResponseMember::to_owned)
76 .collect(),
77 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
78 }
79 }
80 fn encode_field_0<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
81 if version >= 2 {
82 put_i32(buf, self.throttle_time_ms);
83 }
84 }
85 fn encode_field_1<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
86 if version >= 0 {
87 put_i16(buf, self.error_code);
88 }
89 }
90 fn encode_field_2<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
91 if version >= 0 {
92 put_i32(buf, self.generation_id);
93 }
94 }
95 fn encode_field_3<B: BufMut>(&self, buf: &mut B, version: i16, flex: bool) {
96 if version >= 7 {
97 if flex {
98 let () = put_compact_nullable_string(buf, self.protocol_type);
99 } else {
100 let () = put_nullable_string(buf, self.protocol_type);
101 }
102 }
103 }
104 fn encode_field_4<B: BufMut>(&self, buf: &mut B, version: i16, flex: bool) {
105 if version >= 0 {
106 if version >= 7 {
107 if flex {
108 let () = put_compact_nullable_string(buf, self.protocol_name);
109 } else {
110 let () = put_nullable_string(buf, self.protocol_name);
111 }
112 } else {
113 if flex {
114 let () = put_compact_string(buf, (self.protocol_name).unwrap_or(""));
115 } else {
116 let () = put_string(buf, (self.protocol_name).unwrap_or(""));
117 }
118 }
119 }
120 }
121 fn encode_field_5<B: BufMut>(&self, buf: &mut B, version: i16, flex: bool) {
122 if version >= 0 {
123 if flex {
124 let () = put_compact_string(buf, self.leader);
125 } else {
126 let () = put_string(buf, self.leader);
127 }
128 }
129 }
130 fn encode_field_6<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
131 if version >= 9 {
132 put_bool(buf, self.skip_assignment);
133 }
134 }
135 fn encode_field_7<B: BufMut>(&self, buf: &mut B, version: i16, flex: bool) {
136 if version >= 0 {
137 if flex {
138 let () = put_compact_string(buf, self.member_id);
139 } else {
140 let () = put_string(buf, self.member_id);
141 }
142 }
143 }
144 fn encode_field_8<B: BufMut>(
145 &self,
146 buf: &mut B,
147 version: i16,
148 flex: bool,
149 ) -> Result<(), ProtocolError> {
150 if version >= 0 {
151 {
152 crate::primitives::array::put_array_len(buf, (self.members).len(), flex);
153 for it in &self.members {
154 it.encode(buf, version)?;
155 }
156 };
157 }
158 Ok(())
159 }
160 fn encode_tagged_fields<B: BufMut>(&self, buf: &mut B, _version: i16, flex: bool) {
161 if flex {
162 let tagged = WriteTaggedFields::new();
163 tagged.write(buf, &self.unknown_tagged_fields);
164 }
165 }
166 fn decode_field_0(
167 out: &mut Self,
168 buf: &mut &'a [u8],
169 version: i16,
170 _flex: bool,
171 ) -> Result<(), ProtocolError> {
172 if version >= 2 {
173 out.throttle_time_ms = get_i32(buf)?;
174 }
175 Ok(())
176 }
177 fn decode_field_1(
178 out: &mut Self,
179 buf: &mut &'a [u8],
180 version: i16,
181 _flex: bool,
182 ) -> Result<(), ProtocolError> {
183 if version >= 0 {
184 out.error_code = get_i16(buf)?;
185 }
186 Ok(())
187 }
188 fn decode_field_2(
189 out: &mut Self,
190 buf: &mut &'a [u8],
191 version: i16,
192 _flex: bool,
193 ) -> Result<(), ProtocolError> {
194 if version >= 0 {
195 out.generation_id = get_i32(buf)?;
196 }
197 Ok(())
198 }
199 fn decode_field_3(
200 out: &mut Self,
201 buf: &mut &'a [u8],
202 version: i16,
203 flex: bool,
204 ) -> Result<(), ProtocolError> {
205 if version >= 7 {
206 out.protocol_type = if flex {
207 get_compact_nullable_string_borrowed(buf)?
208 } else {
209 get_nullable_string_borrowed(buf)?
210 };
211 }
212 Ok(())
213 }
214 fn decode_field_4(
215 out: &mut Self,
216 buf: &mut &'a [u8],
217 version: i16,
218 flex: bool,
219 ) -> Result<(), ProtocolError> {
220 if version >= 0 {
221 out.protocol_name = if version >= 7 {
222 if flex {
223 get_compact_nullable_string_borrowed(buf)?
224 } else {
225 get_nullable_string_borrowed(buf)?
226 }
227 } else {
228 Some(if flex {
229 get_compact_string_borrowed(buf)?
230 } else {
231 get_string_borrowed(buf)?
232 })
233 };
234 }
235 Ok(())
236 }
237 fn decode_field_5(
238 out: &mut Self,
239 buf: &mut &'a [u8],
240 version: i16,
241 flex: bool,
242 ) -> Result<(), ProtocolError> {
243 if version >= 0 {
244 out.leader = if flex {
245 get_compact_string_borrowed(buf)?
246 } else {
247 get_string_borrowed(buf)?
248 };
249 }
250 Ok(())
251 }
252 fn decode_field_6(
253 out: &mut Self,
254 buf: &mut &'a [u8],
255 version: i16,
256 _flex: bool,
257 ) -> Result<(), ProtocolError> {
258 if version >= 9 {
259 out.skip_assignment = get_bool(buf)?;
260 }
261 Ok(())
262 }
263 fn decode_field_7(
264 out: &mut Self,
265 buf: &mut &'a [u8],
266 version: i16,
267 flex: bool,
268 ) -> Result<(), ProtocolError> {
269 if version >= 0 {
270 out.member_id = if flex {
271 get_compact_string_borrowed(buf)?
272 } else {
273 get_string_borrowed(buf)?
274 };
275 }
276 Ok(())
277 }
278 fn decode_field_8(
279 out: &mut Self,
280 buf: &mut &'a [u8],
281 version: i16,
282 flex: bool,
283 ) -> Result<(), ProtocolError> {
284 if version >= 0 {
285 out.members = {
286 let n = crate::primitives::array::get_array_len(buf, flex)?;
287 let mut v = Vec::with_capacity(n);
288 for _ in 0..n {
289 v.push(JoinGroupResponseMember::decode_borrow(buf, version)?);
290 }
291 v
292 };
293 }
294 Ok(())
295 }
296 fn decode_tagged_fields(
297 out: &mut Self,
298 buf: &mut &'a [u8],
299 _version: i16,
300 flex: bool,
301 ) -> Result<(), ProtocolError> {
302 if flex {
303 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
304 }
305 Ok(())
306 }
307}
308impl Encode for JoinGroupResponse<'_> {
309 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
310 if !(MIN_VERSION..=MAX_VERSION).contains(&version) {
311 return Err(ProtocolError::UnsupportedVersion {
312 api_key: API_KEY,
313 version,
314 });
315 }
316 let flex = is_flexible(version);
317 self.encode_field_0(buf, version, flex);
318 self.encode_field_1(buf, version, flex);
319 self.encode_field_2(buf, version, flex);
320 self.encode_field_3(buf, version, flex);
321 self.encode_field_4(buf, version, flex);
322 self.encode_field_5(buf, version, flex);
323 self.encode_field_6(buf, version, flex);
324 self.encode_field_7(buf, version, flex);
325 self.encode_field_8(buf, version, flex)?;
326 self.encode_tagged_fields(buf, version, flex);
327 Ok(())
328 }
329 fn encoded_len(&self, version: i16) -> usize {
330 let flex = is_flexible(version);
331 let mut n: usize = 0;
332 if version >= 2 {
333 n += 4;
334 }
335 if version >= 0 {
336 n += 2;
337 }
338 if version >= 0 {
339 n += 4;
340 }
341 if version >= 7 {
342 n += if flex {
343 compact_nullable_string_len(self.protocol_type)
344 } else {
345 nullable_string_len(self.protocol_type)
346 };
347 }
348 if version >= 0 {
349 n += if version >= 7 {
350 if flex {
351 compact_nullable_string_len(self.protocol_name)
352 } else {
353 nullable_string_len(self.protocol_name)
354 }
355 } else {
356 if flex {
357 compact_string_len((self.protocol_name).unwrap_or(""))
358 } else {
359 string_len((self.protocol_name).unwrap_or(""))
360 }
361 };
362 }
363 if version >= 0 {
364 n += if flex {
365 compact_string_len(self.leader)
366 } else {
367 string_len(self.leader)
368 };
369 }
370 if version >= 9 {
371 n += 1;
372 }
373 if version >= 0 {
374 n += if flex {
375 compact_string_len(self.member_id)
376 } else {
377 string_len(self.member_id)
378 };
379 }
380 if version >= 0 {
381 n += {
382 let prefix =
383 crate::primitives::array::array_len_prefix_len((self.members).len(), flex);
384 let body: usize = (self.members)
385 .iter()
386 .map(|it| it.encoded_len(version))
387 .sum();
388 prefix + body
389 };
390 }
391 if flex {
392 let known_pairs: Vec<(u32, usize)> = Vec::new();
393 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
394 }
395 n
396 }
397}
398impl<'de> DecodeBorrow<'de> for JoinGroupResponse<'de> {
399 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
400 if !(MIN_VERSION..=MAX_VERSION).contains(&version) {
401 return Err(ProtocolError::UnsupportedVersion {
402 api_key: API_KEY,
403 version,
404 });
405 }
406 let flex = is_flexible(version);
407 let mut out = Self::default();
408 Self::decode_field_0(&mut out, buf, version, flex)?;
409 Self::decode_field_1(&mut out, buf, version, flex)?;
410 Self::decode_field_2(&mut out, buf, version, flex)?;
411 Self::decode_field_3(&mut out, buf, version, flex)?;
412 Self::decode_field_4(&mut out, buf, version, flex)?;
413 Self::decode_field_5(&mut out, buf, version, flex)?;
414 Self::decode_field_6(&mut out, buf, version, flex)?;
415 Self::decode_field_7(&mut out, buf, version, flex)?;
416 Self::decode_field_8(&mut out, buf, version, flex)?;
417 Self::decode_tagged_fields(&mut out, buf, version, flex)?;
418 Ok(out)
419 }
420}
421#[cfg(test)]
422impl JoinGroupResponse<'_> {
423 #[must_use]
424 pub fn populated(version: i16) -> Self {
425 let mut m = Self::default();
426 if version >= 2 {
427 m.throttle_time_ms = 1i32;
428 }
429 if version >= 0 {
430 m.error_code = 1i16;
431 }
432 if version >= 0 {
433 m.generation_id = 1i32;
434 }
435 if version >= 7 {
436 m.protocol_type = Some("x");
437 }
438 if version >= 0 {
439 m.protocol_name = Some("x");
440 }
441 if version >= 0 {
442 m.leader = "x";
443 }
444 if version >= 9 {
445 m.skip_assignment = true;
446 }
447 if version >= 0 {
448 m.member_id = "x";
449 }
450 if version >= 0 {
451 m.members = vec![JoinGroupResponseMember::populated(version)];
452 }
453 m
454 }
455}
456#[derive(Debug, Clone, PartialEq, Eq, Default)]
457pub struct JoinGroupResponseMember<'a> {
458 pub member_id: &'a str,
459 pub group_instance_id: Option<&'a str>,
460 pub metadata: &'a [u8],
461 pub unknown_tagged_fields: UnknownTaggedFields,
462}
463impl JoinGroupResponseMember<'_> {
464 #[must_use]
468 pub fn to_owned(&self) -> crate::owned::join_group_response::JoinGroupResponseMember {
469 crate::owned::join_group_response::JoinGroupResponseMember {
470 member_id: (self.member_id).to_string(),
471 group_instance_id: (self.group_instance_id).map(std::string::ToString::to_string),
472 metadata: Bytes::copy_from_slice(self.metadata),
473 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
474 }
475 }
476}
477impl Encode for JoinGroupResponseMember<'_> {
478 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
479 let flex = version >= 6;
480 if version >= 0 {
481 if flex {
482 let () = put_compact_string(buf, self.member_id);
483 } else {
484 let () = put_string(buf, self.member_id);
485 }
486 }
487 if version >= 5 {
488 if flex {
489 let () = put_compact_nullable_string(buf, self.group_instance_id);
490 } else {
491 let () = put_nullable_string(buf, self.group_instance_id);
492 }
493 }
494 if version >= 0 {
495 if flex {
496 let () = put_compact_bytes(buf, self.metadata);
497 } else {
498 let () = put_bytes(buf, self.metadata);
499 }
500 }
501 if flex {
502 let tagged = WriteTaggedFields::new();
503 tagged.write(buf, &self.unknown_tagged_fields);
504 }
505 Ok(())
506 }
507 fn encoded_len(&self, version: i16) -> usize {
508 let flex = version >= 6;
509 let mut n: usize = 0;
510 if version >= 0 {
511 n += if flex {
512 compact_string_len(self.member_id)
513 } else {
514 string_len(self.member_id)
515 };
516 }
517 if version >= 5 {
518 n += if flex {
519 compact_nullable_string_len(self.group_instance_id)
520 } else {
521 nullable_string_len(self.group_instance_id)
522 };
523 }
524 if version >= 0 {
525 n += if flex {
526 crate::primitives::varint::uvarint_len(
527 u32::try_from((self.metadata).len() + 1).unwrap(),
528 ) + (self.metadata).len()
529 } else {
530 4 + (self.metadata).len()
531 };
532 }
533 if flex {
534 let known_pairs: Vec<(u32, usize)> = Vec::new();
535 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
536 }
537 n
538 }
539}
540impl<'de> DecodeBorrow<'de> for JoinGroupResponseMember<'de> {
541 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
542 let flex = version >= 6;
543 let mut out = Self::default();
544 if version >= 0 {
545 out.member_id = if flex {
546 get_compact_string_borrowed(buf)?
547 } else {
548 get_string_borrowed(buf)?
549 };
550 }
551 if version >= 5 {
552 out.group_instance_id = if flex {
553 get_compact_nullable_string_borrowed(buf)?
554 } else {
555 get_nullable_string_borrowed(buf)?
556 };
557 }
558 if version >= 0 {
559 out.metadata = if flex {
560 get_compact_bytes_borrowed(buf)?
561 } else {
562 get_bytes_borrowed(buf)?
563 };
564 }
565 if flex {
566 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
567 }
568 Ok(out)
569 }
570}
571#[cfg(test)]
572impl JoinGroupResponseMember<'_> {
573 #[must_use]
574 pub fn populated(version: i16) -> Self {
575 let mut m = Self::default();
576 if version >= 0 {
577 m.member_id = "x";
578 }
579 if version >= 5 {
580 m.group_instance_id = Some("x");
581 }
582 if version >= 0 {
583 m.metadata = &b"x"[..];
584 }
585 m
586 }
587}