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