kafrust_protocol/api/
describe_groups.rs

1use crate::codec::{Decoder, Encoder};
2use crate::error::Result;
3use crate::header::RequestHeader;
4
5pub const API_KEY: i16 = 15;
6
7#[derive(Debug, Clone, PartialEq, Eq)]
8pub struct DescribeGroupsRequestV1 {
9    pub correlation_id: i32,
10    pub client_id: Option<String>,
11    pub group_ids: Vec<String>,
12}
13
14impl DescribeGroupsRequestV1 {
15    pub fn encode(&self) -> Result<Vec<u8>> {
16        let mut encoder = Encoder::new();
17        RequestHeader {
18            api_key: API_KEY,
19            api_version: 1,
20            correlation_id: self.correlation_id,
21            client_id: self.client_id.clone(),
22        }
23        .encode_v1(&mut encoder)?;
24        encoder.write_array(Some(&self.group_ids), |encoder, group_id| {
25            encoder.write_string(group_id)
26        })?;
27        Ok(encoder.into_bytes())
28    }
29}
30
31#[derive(Debug, Clone, PartialEq, Eq)]
32pub struct DescribeGroupsResponseV1 {
33    pub throttle_time_ms: i32,
34    pub groups: Vec<DescribeGroupsGroupV1>,
35}
36
37impl DescribeGroupsResponseV1 {
38    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
39        Ok(Self {
40            throttle_time_ms: decoder.read_i32()?,
41            groups: decoder
42                .read_array("describe groups", DescribeGroupsGroupV1::decode)?
43                .unwrap_or_default(),
44        })
45    }
46}
47
48#[derive(Debug, Clone, PartialEq, Eq)]
49pub struct DescribeGroupsGroupV1 {
50    pub error_code: i16,
51    pub group_id: String,
52    pub state: String,
53    pub protocol_type: String,
54    pub protocol_data: String,
55    pub members: Vec<DescribeGroupsMemberV1>,
56}
57
58impl DescribeGroupsGroupV1 {
59    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
60        Ok(Self {
61            error_code: decoder.read_i16()?,
62            group_id: decoder.read_string()?,
63            state: decoder.read_string()?,
64            protocol_type: decoder.read_string()?,
65            protocol_data: decoder.read_string()?,
66            members: decoder
67                .read_array("describe group members", DescribeGroupsMemberV1::decode)?
68                .unwrap_or_default(),
69        })
70    }
71}
72
73#[derive(Debug, Clone, PartialEq, Eq)]
74pub struct DescribeGroupsMemberV1 {
75    pub member_id: String,
76    pub client_id: String,
77    pub client_host: String,
78    pub member_metadata: Vec<u8>,
79    pub member_assignment: Vec<u8>,
80}
81
82impl DescribeGroupsMemberV1 {
83    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
84        Ok(Self {
85            member_id: decoder.read_string()?,
86            client_id: decoder.read_string()?,
87            client_host: decoder.read_string()?,
88            member_metadata: decoder.read_bytes()?,
89            member_assignment: decoder.read_bytes()?,
90        })
91    }
92}
93
94#[cfg(test)]
95#[allow(clippy::unwrap_used)]
96mod tests {
97    use super::{DescribeGroupsRequestV1, DescribeGroupsResponseV1, API_KEY};
98    use crate::codec::Decoder;
99
100    #[test]
101    fn encodes_describe_groups_v1_request() {
102        let request = DescribeGroupsRequestV1 {
103            correlation_id: 14,
104            client_id: Some("kafrust".to_owned()),
105            group_ids: vec!["orders-group".to_owned()],
106        };
107
108        assert_eq!(
109            request.encode().unwrap(),
110            [
111                0, 15, // API key
112                0, 1, // API version
113                0, 0, 0, 14, // correlation ID
114                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // client ID
115                0, 0, 0, 1, // group count
116                0, 12, b'o', b'r', b'd', b'e', b'r', b's', b'-', b'g', b'r', b'o', b'u', b'p',
117            ]
118        );
119        assert_eq!(API_KEY, 15);
120    }
121
122    #[test]
123    fn decodes_describe_groups_v1_response() {
124        let bytes = [
125            0, 0, 0, 4, // throttle time
126            0, 0, 0, 1, // group count
127            0, 0, // success
128            0, 12, b'o', b'r', b'd', b'e', b'r', b's', b'-', b'g', b'r', b'o', b'u', b'p', 0, 6,
129            b'S', b't', b'a', b'b', b'l', b'e', // state
130            0, 8, b'c', b'o', b'n', b's', b'u', b'm', b'e', b'r', // protocol type
131            0, 5, b'r', b'a', b'n', b'g', b'e', // protocol
132            0, 0, 0, 1, // member count
133            0, 8, b'm', b'e', b'm', b'b', b'e', b'r', b'-', b'1', // member ID
134            0, 8, b'c', b'l', b'i', b'e', b'n', b't', b'-', b'1', // client ID
135            0, 10, b'/', b'1', b'2', b'7', b'.', b'0', b'.', b'0', b'.', b'1', // client host
136            0, 0, 0, 2, 1, 2, // member metadata
137            0, 0, 0, 3, 3, 4, 5, // member assignment
138        ];
139        let mut decoder = Decoder::new(&bytes);
140
141        let response = DescribeGroupsResponseV1::decode_body(&mut decoder).unwrap();
142
143        assert_eq!(response.throttle_time_ms, 4);
144        assert_eq!(response.groups.len(), 1);
145        assert_eq!(response.groups[0].group_id, "orders-group");
146        assert_eq!(response.groups[0].state, "Stable");
147        assert_eq!(response.groups[0].protocol_type, "consumer");
148        assert_eq!(response.groups[0].protocol_data, "range");
149        assert_eq!(response.groups[0].members.len(), 1);
150        assert_eq!(response.groups[0].members[0].client_host, "/127.0.0.1");
151        assert_eq!(response.groups[0].members[0].member_metadata, [1, 2]);
152        assert_eq!(response.groups[0].members[0].member_assignment, [3, 4, 5]);
153        assert!(decoder.is_empty());
154    }
155}
kafrust_protocol/api/describe_groups.rs

kafrust_protocol/api/
describe_groups.rs
