iggy_binary_protocol/responses/consumer_groups/
get_consumer_groups.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use super::consumer_group_response::ConsumerGroupResponse;
19use crate::WireError;
20use crate::codec::{WireDecode, WireEncode};
21use bytes::BytesMut;
22
23/// `GetConsumerGroups` response: sequential consumer group headers.
24///
25/// Wire format:
26/// ```text
27/// [ConsumerGroupResponse]*
28/// ```
29///
30/// Empty payload means zero groups.
31#[derive(Debug, Clone, PartialEq, Eq)]
32pub struct GetConsumerGroupsResponse {
33    pub groups: Vec<ConsumerGroupResponse>,
34}
35
36impl WireEncode for GetConsumerGroupsResponse {
37    fn encoded_size(&self) -> usize {
38        self.groups.iter().map(WireEncode::encoded_size).sum()
39    }
40
41    fn encode(&self, buf: &mut BytesMut) {
42        for group in &self.groups {
43            group.encode(buf);
44        }
45    }
46}
47
48impl WireDecode for GetConsumerGroupsResponse {
49    fn decode(buf: &[u8]) -> Result<(Self, usize), WireError> {
50        let mut groups = Vec::new();
51        let mut pos = 0;
52        while pos < buf.len() {
53            let (group, consumed) = ConsumerGroupResponse::decode(&buf[pos..])?;
54            pos += consumed;
55            groups.push(group);
56        }
57        Ok((Self { groups }, pos))
58    }
59}
60
61#[cfg(test)]
62mod tests {
63    use super::*;
64    use crate::WireName;
65
66    fn sample_group(id: u32, name: &str) -> ConsumerGroupResponse {
67        ConsumerGroupResponse {
68            id,
69            partitions_count: 4,
70            members_count: 2,
71            name: WireName::new(name).unwrap(),
72        }
73    }
74
75    #[test]
76    fn roundtrip_empty() {
77        let resp = GetConsumerGroupsResponse { groups: vec![] };
78        let bytes = resp.to_bytes();
79        assert!(bytes.is_empty());
80        let (decoded, consumed) = GetConsumerGroupsResponse::decode(&bytes).unwrap();
81        assert_eq!(consumed, 0);
82        assert_eq!(decoded, resp);
83    }
84
85    #[test]
86    fn roundtrip_multiple() {
87        let resp = GetConsumerGroupsResponse {
88            groups: vec![sample_group(1, "group-a"), sample_group(2, "group-b")],
89        };
90        let bytes = resp.to_bytes();
91        let (decoded, consumed) = GetConsumerGroupsResponse::decode(&bytes).unwrap();
92        assert_eq!(consumed, bytes.len());
93        assert_eq!(decoded, resp);
94    }
95
96    #[test]
97    fn truncated_returns_error() {
98        let resp = GetConsumerGroupsResponse {
99            groups: vec![sample_group(1, "g")],
100        };
101        let bytes = resp.to_bytes();
102        for i in 1..bytes.len() {
103            assert!(
104                GetConsumerGroupsResponse::decode(&bytes[..i]).is_err(),
105                "expected error for truncation at byte {i}"
106            );
107        }
108    }
109}
iggy_binary_protocol/responses/consumer_groups/get_consumer_groups.rs

iggy_binary_protocol/responses/consumer_groups/
get_consumer_groups.rs
