1use bytes::{Buf, BufMut};
4
5use crate::{
6 Decode, Encode, ProtocolError, UnknownTaggedFields,
7 primitives::{
8 fixed::{get_i16, get_i32, get_i64, put_i16, put_i32, put_i64},
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 FLEXIBLE_MIN: i16 = 6;
22#[inline]
23#[must_use]
24pub fn is_flexible(version: i16) -> bool {
25 version >= FLEXIBLE_MIN
26}
27#[derive(Debug, Clone, PartialEq, Eq, Default)]
28pub struct OffsetFetchResponse {
29 pub throttle_time_ms: i32,
30 pub topics: Vec<OffsetFetchResponseTopic>,
31 pub error_code: i16,
32 pub groups: Vec<OffsetFetchResponseGroup>,
33 pub unknown_tagged_fields: UnknownTaggedFields,
34}
35impl Encode for OffsetFetchResponse {
36 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
37 if !(MIN_VERSION..=MAX_VERSION).contains(&version) {
38 return Err(ProtocolError::UnsupportedVersion {
39 api_key: API_KEY,
40 version,
41 });
42 }
43 let flex = is_flexible(version);
44 if version >= 3 {
45 put_i32(buf, self.throttle_time_ms);
46 }
47 if (0..=7).contains(&version) {
48 {
49 crate::primitives::array::put_array_len(buf, (self.topics).len(), flex);
50 for it in &self.topics {
51 it.encode(buf, version)?;
52 }
53 };
54 }
55 if (2..=7).contains(&version) {
56 put_i16(buf, self.error_code);
57 }
58 if version >= 8 {
59 {
60 crate::primitives::array::put_array_len(buf, (self.groups).len(), flex);
61 for it in &self.groups {
62 it.encode(buf, version)?;
63 }
64 };
65 }
66 if flex {
67 let tagged = WriteTaggedFields::new();
68 tagged.write(buf, &self.unknown_tagged_fields);
69 }
70 Ok(())
71 }
72 fn encoded_len(&self, version: i16) -> usize {
73 let flex = is_flexible(version);
74 let mut n: usize = 0;
75 if version >= 3 {
76 n += 4;
77 }
78 if (0..=7).contains(&version) {
79 n += {
80 let prefix =
81 crate::primitives::array::array_len_prefix_len((self.topics).len(), flex);
82 let body: usize = (self.topics).iter().map(|it| it.encoded_len(version)).sum();
83 prefix + body
84 };
85 }
86 if (2..=7).contains(&version) {
87 n += 2;
88 }
89 if version >= 8 {
90 n += {
91 let prefix =
92 crate::primitives::array::array_len_prefix_len((self.groups).len(), flex);
93 let body: usize = (self.groups).iter().map(|it| it.encoded_len(version)).sum();
94 prefix + body
95 };
96 }
97 if flex {
98 let known_pairs: Vec<(u32, usize)> = Vec::new();
99 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
100 }
101 n
102 }
103}
104impl Decode<'_> for OffsetFetchResponse {
105 fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
106 if !(MIN_VERSION..=MAX_VERSION).contains(&version) {
107 return Err(ProtocolError::UnsupportedVersion {
108 api_key: API_KEY,
109 version,
110 });
111 }
112 let flex = is_flexible(version);
113 let mut out = Self::default();
114 if version >= 3 {
115 out.throttle_time_ms = get_i32(buf)?;
116 }
117 if (0..=7).contains(&version) {
118 out.topics = {
119 let n = crate::primitives::array::get_array_len(buf, flex)?;
120 let mut v = Vec::with_capacity(n);
121 for _ in 0..n {
122 v.push(OffsetFetchResponseTopic::decode(buf, version)?);
123 }
124 v
125 };
126 }
127 if (2..=7).contains(&version) {
128 out.error_code = get_i16(buf)?;
129 }
130 if version >= 8 {
131 out.groups = {
132 let n = crate::primitives::array::get_array_len(buf, flex)?;
133 let mut v = Vec::with_capacity(n);
134 for _ in 0..n {
135 v.push(OffsetFetchResponseGroup::decode(buf, version)?);
136 }
137 v
138 };
139 }
140 if flex {
141 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
142 }
143 Ok(out)
144 }
145}
146#[cfg(test)]
147impl OffsetFetchResponse {
148 #[must_use]
149 pub fn populated(version: i16) -> Self {
150 let mut m = Self::default();
151 if version >= 3 {
152 m.throttle_time_ms = 1i32;
153 }
154 if (0..=7).contains(&version) {
155 m.topics = vec![OffsetFetchResponseTopic::populated(version)];
156 }
157 if (2..=7).contains(&version) {
158 m.error_code = 1i16;
159 }
160 if version >= 8 {
161 m.groups = vec![OffsetFetchResponseGroup::populated(version)];
162 }
163 m
164 }
165}
166#[derive(Debug, Clone, PartialEq, Eq, Default)]
167pub struct OffsetFetchResponseTopic {
168 pub name: String,
169 pub partitions: Vec<OffsetFetchResponsePartition>,
170 pub unknown_tagged_fields: UnknownTaggedFields,
171}
172impl Encode for OffsetFetchResponseTopic {
173 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
174 let flex = version >= 6;
175 if (0..=7).contains(&version) {
176 if flex {
177 let () = put_compact_string(buf, &self.name);
178 } else {
179 let () = put_string(buf, &self.name);
180 }
181 }
182 if (0..=7).contains(&version) {
183 {
184 crate::primitives::array::put_array_len(buf, (self.partitions).len(), flex);
185 for it in &self.partitions {
186 it.encode(buf, version)?;
187 }
188 };
189 }
190 if flex {
191 let tagged = WriteTaggedFields::new();
192 tagged.write(buf, &self.unknown_tagged_fields);
193 }
194 Ok(())
195 }
196 fn encoded_len(&self, version: i16) -> usize {
197 let flex = version >= 6;
198 let mut n: usize = 0;
199 if (0..=7).contains(&version) {
200 n += if flex {
201 compact_string_len(&self.name)
202 } else {
203 string_len(&self.name)
204 };
205 }
206 if (0..=7).contains(&version) {
207 n += {
208 let prefix =
209 crate::primitives::array::array_len_prefix_len((self.partitions).len(), flex);
210 let body: usize = (self.partitions)
211 .iter()
212 .map(|it| it.encoded_len(version))
213 .sum();
214 prefix + body
215 };
216 }
217 if flex {
218 let known_pairs: Vec<(u32, usize)> = Vec::new();
219 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
220 }
221 n
222 }
223}
224impl Decode<'_> for OffsetFetchResponseTopic {
225 fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
226 let flex = version >= 6;
227 let mut out = Self::default();
228 if (0..=7).contains(&version) {
229 out.name = if flex {
230 get_compact_string_owned(buf)?
231 } else {
232 get_string_owned(buf)?
233 };
234 }
235 if (0..=7).contains(&version) {
236 out.partitions = {
237 let n = crate::primitives::array::get_array_len(buf, flex)?;
238 let mut v = Vec::with_capacity(n);
239 for _ in 0..n {
240 v.push(OffsetFetchResponsePartition::decode(buf, version)?);
241 }
242 v
243 };
244 }
245 if flex {
246 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
247 }
248 Ok(out)
249 }
250}
251#[cfg(test)]
252impl OffsetFetchResponseTopic {
253 #[must_use]
254 pub fn populated(version: i16) -> Self {
255 let mut m = Self::default();
256 if (0..=7).contains(&version) {
257 m.name = "x".to_string();
258 }
259 if (0..=7).contains(&version) {
260 m.partitions = vec![OffsetFetchResponsePartition::populated(version)];
261 }
262 m
263 }
264}
265#[derive(Debug, Clone, PartialEq, Eq)]
266pub struct OffsetFetchResponsePartition {
267 pub partition_index: i32,
268 pub committed_offset: i64,
269 pub committed_leader_epoch: i32,
270 pub metadata: Option<String>,
271 pub error_code: i16,
272 pub unknown_tagged_fields: UnknownTaggedFields,
273}
274impl Default for OffsetFetchResponsePartition {
275 fn default() -> Self {
276 Self {
277 partition_index: 0i32,
278 committed_offset: 0i64,
279 committed_leader_epoch: -1i32,
280 metadata: Some(String::new()),
281 error_code: 0i16,
282 unknown_tagged_fields: UnknownTaggedFields::default(),
283 }
284 }
285}
286impl Encode for OffsetFetchResponsePartition {
287 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
288 let flex = version >= 6;
289 if (0..=7).contains(&version) {
290 put_i32(buf, self.partition_index);
291 }
292 if (0..=7).contains(&version) {
293 put_i64(buf, self.committed_offset);
294 }
295 if (5..=7).contains(&version) {
296 put_i32(buf, self.committed_leader_epoch);
297 }
298 if (0..=7).contains(&version) {
299 if flex {
300 let () = put_compact_nullable_string(buf, self.metadata.as_deref());
301 } else {
302 let () = put_nullable_string(buf, self.metadata.as_deref());
303 }
304 }
305 if (0..=7).contains(&version) {
306 put_i16(buf, self.error_code);
307 }
308 if flex {
309 let tagged = WriteTaggedFields::new();
310 tagged.write(buf, &self.unknown_tagged_fields);
311 }
312 Ok(())
313 }
314 fn encoded_len(&self, version: i16) -> usize {
315 let flex = version >= 6;
316 let mut n: usize = 0;
317 if (0..=7).contains(&version) {
318 n += 4;
319 }
320 if (0..=7).contains(&version) {
321 n += 8;
322 }
323 if (5..=7).contains(&version) {
324 n += 4;
325 }
326 if (0..=7).contains(&version) {
327 n += if flex {
328 compact_nullable_string_len(self.metadata.as_deref())
329 } else {
330 nullable_string_len(self.metadata.as_deref())
331 };
332 }
333 if (0..=7).contains(&version) {
334 n += 2;
335 }
336 if flex {
337 let known_pairs: Vec<(u32, usize)> = Vec::new();
338 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
339 }
340 n
341 }
342}
343impl Decode<'_> for OffsetFetchResponsePartition {
344 fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
345 let flex = version >= 6;
346 let mut out = Self::default();
347 if (0..=7).contains(&version) {
348 out.partition_index = get_i32(buf)?;
349 }
350 if (0..=7).contains(&version) {
351 out.committed_offset = get_i64(buf)?;
352 }
353 if (5..=7).contains(&version) {
354 out.committed_leader_epoch = get_i32(buf)?;
355 }
356 if (0..=7).contains(&version) {
357 out.metadata = if flex {
358 get_compact_nullable_string_owned(buf)?
359 } else {
360 get_nullable_string_owned(buf)?
361 };
362 }
363 if (0..=7).contains(&version) {
364 out.error_code = get_i16(buf)?;
365 }
366 if flex {
367 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
368 }
369 Ok(out)
370 }
371}
372#[cfg(test)]
373impl OffsetFetchResponsePartition {
374 #[must_use]
375 pub fn populated(version: i16) -> Self {
376 let mut m = Self::default();
377 if (0..=7).contains(&version) {
378 m.partition_index = 1i32;
379 }
380 if (0..=7).contains(&version) {
381 m.committed_offset = 1i64;
382 }
383 if (5..=7).contains(&version) {
384 m.committed_leader_epoch = 1i32;
385 }
386 if (0..=7).contains(&version) {
387 m.metadata = Some("x".to_string());
388 }
389 if (0..=7).contains(&version) {
390 m.error_code = 1i16;
391 }
392 m
393 }
394}
395#[derive(Debug, Clone, PartialEq, Eq, Default)]
396pub struct OffsetFetchResponseGroup {
397 pub group_id: String,
398 pub topics: Vec<OffsetFetchResponseTopics>,
399 pub error_code: i16,
400 pub unknown_tagged_fields: UnknownTaggedFields,
401}
402impl Encode for OffsetFetchResponseGroup {
403 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
404 let flex = version >= 6;
405 if version >= 8 {
406 if flex {
407 let () = put_compact_string(buf, &self.group_id);
408 } else {
409 let () = put_string(buf, &self.group_id);
410 }
411 }
412 if version >= 8 {
413 {
414 crate::primitives::array::put_array_len(buf, (self.topics).len(), flex);
415 for it in &self.topics {
416 it.encode(buf, version)?;
417 }
418 };
419 }
420 if version >= 8 {
421 put_i16(buf, self.error_code);
422 }
423 if flex {
424 let tagged = WriteTaggedFields::new();
425 tagged.write(buf, &self.unknown_tagged_fields);
426 }
427 Ok(())
428 }
429 fn encoded_len(&self, version: i16) -> usize {
430 let flex = version >= 6;
431 let mut n: usize = 0;
432 if version >= 8 {
433 n += if flex {
434 compact_string_len(&self.group_id)
435 } else {
436 string_len(&self.group_id)
437 };
438 }
439 if version >= 8 {
440 n += {
441 let prefix =
442 crate::primitives::array::array_len_prefix_len((self.topics).len(), flex);
443 let body: usize = (self.topics).iter().map(|it| it.encoded_len(version)).sum();
444 prefix + body
445 };
446 }
447 if version >= 8 {
448 n += 2;
449 }
450 if flex {
451 let known_pairs: Vec<(u32, usize)> = Vec::new();
452 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
453 }
454 n
455 }
456}
457impl Decode<'_> for OffsetFetchResponseGroup {
458 fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
459 let flex = version >= 6;
460 let mut out = Self::default();
461 if version >= 8 {
462 out.group_id = if flex {
463 get_compact_string_owned(buf)?
464 } else {
465 get_string_owned(buf)?
466 };
467 }
468 if version >= 8 {
469 out.topics = {
470 let n = crate::primitives::array::get_array_len(buf, flex)?;
471 let mut v = Vec::with_capacity(n);
472 for _ in 0..n {
473 v.push(OffsetFetchResponseTopics::decode(buf, version)?);
474 }
475 v
476 };
477 }
478 if version >= 8 {
479 out.error_code = get_i16(buf)?;
480 }
481 if flex {
482 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
483 }
484 Ok(out)
485 }
486}
487#[cfg(test)]
488impl OffsetFetchResponseGroup {
489 #[must_use]
490 pub fn populated(version: i16) -> Self {
491 let mut m = Self::default();
492 if version >= 8 {
493 m.group_id = "x".to_string();
494 }
495 if version >= 8 {
496 m.topics = vec![OffsetFetchResponseTopics::populated(version)];
497 }
498 if version >= 8 {
499 m.error_code = 1i16;
500 }
501 m
502 }
503}
504#[derive(Debug, Clone, PartialEq, Eq, Default)]
505pub struct OffsetFetchResponseTopics {
506 pub name: String,
507 pub topic_id: crate::primitives::uuid::Uuid,
508 pub partitions: Vec<OffsetFetchResponsePartitions>,
509 pub unknown_tagged_fields: UnknownTaggedFields,
510}
511impl Encode for OffsetFetchResponseTopics {
512 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
513 let flex = version >= 6;
514 if (8..=9).contains(&version) {
515 if flex {
516 let () = put_compact_string(buf, &self.name);
517 } else {
518 let () = put_string(buf, &self.name);
519 }
520 }
521 if version >= 10 {
522 crate::primitives::uuid::put_uuid(buf, self.topic_id);
523 }
524 if version >= 8 {
525 {
526 crate::primitives::array::put_array_len(buf, (self.partitions).len(), flex);
527 for it in &self.partitions {
528 it.encode(buf, version)?;
529 }
530 };
531 }
532 if flex {
533 let tagged = WriteTaggedFields::new();
534 tagged.write(buf, &self.unknown_tagged_fields);
535 }
536 Ok(())
537 }
538 fn encoded_len(&self, version: i16) -> usize {
539 let flex = version >= 6;
540 let mut n: usize = 0;
541 if (8..=9).contains(&version) {
542 n += if flex {
543 compact_string_len(&self.name)
544 } else {
545 string_len(&self.name)
546 };
547 }
548 if version >= 10 {
549 n += 16;
550 }
551 if version >= 8 {
552 n += {
553 let prefix =
554 crate::primitives::array::array_len_prefix_len((self.partitions).len(), flex);
555 let body: usize = (self.partitions)
556 .iter()
557 .map(|it| it.encoded_len(version))
558 .sum();
559 prefix + body
560 };
561 }
562 if flex {
563 let known_pairs: Vec<(u32, usize)> = Vec::new();
564 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
565 }
566 n
567 }
568}
569impl Decode<'_> for OffsetFetchResponseTopics {
570 fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
571 let flex = version >= 6;
572 let mut out = Self::default();
573 if (8..=9).contains(&version) {
574 out.name = if flex {
575 get_compact_string_owned(buf)?
576 } else {
577 get_string_owned(buf)?
578 };
579 }
580 if version >= 10 {
581 out.topic_id = crate::primitives::uuid::get_uuid(buf)?;
582 }
583 if version >= 8 {
584 out.partitions = {
585 let n = crate::primitives::array::get_array_len(buf, flex)?;
586 let mut v = Vec::with_capacity(n);
587 for _ in 0..n {
588 v.push(OffsetFetchResponsePartitions::decode(buf, version)?);
589 }
590 v
591 };
592 }
593 if flex {
594 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
595 }
596 Ok(out)
597 }
598}
599#[cfg(test)]
600impl OffsetFetchResponseTopics {
601 #[must_use]
602 pub fn populated(version: i16) -> Self {
603 let mut m = Self::default();
604 if (8..=9).contains(&version) {
605 m.name = "x".to_string();
606 }
607 if version >= 10 {
608 m.topic_id = crate::primitives::uuid::Uuid([1u8; 16]);
609 }
610 if version >= 8 {
611 m.partitions = vec![OffsetFetchResponsePartitions::populated(version)];
612 }
613 m
614 }
615}
616#[derive(Debug, Clone, PartialEq, Eq)]
617pub struct OffsetFetchResponsePartitions {
618 pub partition_index: i32,
619 pub committed_offset: i64,
620 pub committed_leader_epoch: i32,
621 pub metadata: Option<String>,
622 pub error_code: i16,
623 pub unknown_tagged_fields: UnknownTaggedFields,
624}
625impl Default for OffsetFetchResponsePartitions {
626 fn default() -> Self {
627 Self {
628 partition_index: 0i32,
629 committed_offset: 0i64,
630 committed_leader_epoch: -1i32,
631 metadata: Some(String::new()),
632 error_code: 0i16,
633 unknown_tagged_fields: UnknownTaggedFields::default(),
634 }
635 }
636}
637impl Encode for OffsetFetchResponsePartitions {
638 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
639 let flex = version >= 6;
640 if version >= 8 {
641 put_i32(buf, self.partition_index);
642 }
643 if version >= 8 {
644 put_i64(buf, self.committed_offset);
645 }
646 if version >= 8 {
647 put_i32(buf, self.committed_leader_epoch);
648 }
649 if version >= 8 {
650 if flex {
651 let () = put_compact_nullable_string(buf, self.metadata.as_deref());
652 } else {
653 let () = put_nullable_string(buf, self.metadata.as_deref());
654 }
655 }
656 if version >= 8 {
657 put_i16(buf, self.error_code);
658 }
659 if flex {
660 let tagged = WriteTaggedFields::new();
661 tagged.write(buf, &self.unknown_tagged_fields);
662 }
663 Ok(())
664 }
665 fn encoded_len(&self, version: i16) -> usize {
666 let flex = version >= 6;
667 let mut n: usize = 0;
668 if version >= 8 {
669 n += 4;
670 }
671 if version >= 8 {
672 n += 8;
673 }
674 if version >= 8 {
675 n += 4;
676 }
677 if version >= 8 {
678 n += if flex {
679 compact_nullable_string_len(self.metadata.as_deref())
680 } else {
681 nullable_string_len(self.metadata.as_deref())
682 };
683 }
684 if version >= 8 {
685 n += 2;
686 }
687 if flex {
688 let known_pairs: Vec<(u32, usize)> = Vec::new();
689 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
690 }
691 n
692 }
693}
694impl Decode<'_> for OffsetFetchResponsePartitions {
695 fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
696 let flex = version >= 6;
697 let mut out = Self::default();
698 if version >= 8 {
699 out.partition_index = get_i32(buf)?;
700 }
701 if version >= 8 {
702 out.committed_offset = get_i64(buf)?;
703 }
704 if version >= 8 {
705 out.committed_leader_epoch = get_i32(buf)?;
706 }
707 if version >= 8 {
708 out.metadata = if flex {
709 get_compact_nullable_string_owned(buf)?
710 } else {
711 get_nullable_string_owned(buf)?
712 };
713 }
714 if version >= 8 {
715 out.error_code = get_i16(buf)?;
716 }
717 if flex {
718 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
719 }
720 Ok(out)
721 }
722}
723#[cfg(test)]
724impl OffsetFetchResponsePartitions {
725 #[must_use]
726 pub fn populated(version: i16) -> Self {
727 let mut m = Self::default();
728 if version >= 8 {
729 m.partition_index = 1i32;
730 }
731 if version >= 8 {
732 m.committed_offset = 1i64;
733 }
734 if version >= 8 {
735 m.committed_leader_epoch = 1i32;
736 }
737 if version >= 8 {
738 m.metadata = Some("x".to_string());
739 }
740 if version >= 8 {
741 m.error_code = 1i16;
742 }
743 m
744 }
745}
746#[must_use]
749#[allow(unused_comparisons)]
750pub fn default_json(version: i16) -> ::serde_json::Value {
751 let mut obj = ::serde_json::Map::new();
752 if version >= 3 {
753 obj.insert("throttleTimeMs".to_string(), ::serde_json::json!(0));
754 }
755 if version <= 7 {
756 obj.insert("topics".to_string(), ::serde_json::Value::Array(vec![]));
757 }
758 if (2..=7).contains(&version) {
759 obj.insert("errorCode".to_string(), ::serde_json::json!(0));
760 }
761 if version >= 8 {
762 obj.insert("groups".to_string(), ::serde_json::Value::Array(vec![]));
763 }
764 ::serde_json::Value::Object(obj)
765}