kafrust_protocol/api/
list_groups.rs

1use crate::codec::{Decoder, Encoder};
2use crate::error::Result;
3use crate::header::RequestHeader;
4
5pub const API_KEY: i16 = 16;
6
7#[derive(Debug, Clone, PartialEq, Eq)]
8pub struct ListGroupsRequestV1 {
9    pub correlation_id: i32,
10    pub client_id: Option<String>,
11}
12
13impl ListGroupsRequestV1 {
14    pub fn encode(&self) -> Result<Vec<u8>> {
15        let mut encoder = Encoder::new();
16        RequestHeader {
17            api_key: API_KEY,
18            api_version: 1,
19            correlation_id: self.correlation_id,
20            client_id: self.client_id.clone(),
21        }
22        .encode_v1(&mut encoder)?;
23        Ok(encoder.into_bytes())
24    }
25}
26
27#[derive(Debug, Clone, PartialEq, Eq)]
28pub struct ListGroupsResponseV1 {
29    pub throttle_time_ms: i32,
30    pub error_code: i16,
31    pub groups: Vec<ListedGroupV1>,
32}
33
34impl ListGroupsResponseV1 {
35    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
36        Ok(Self {
37            throttle_time_ms: decoder.read_i32()?,
38            error_code: decoder.read_i16()?,
39            groups: decoder
40                .read_array("listed groups", ListedGroupV1::decode)?
41                .unwrap_or_default(),
42        })
43    }
44}
45
46#[derive(Debug, Clone, PartialEq, Eq)]
47pub struct ListedGroupV1 {
48    pub group_id: String,
49    pub protocol_type: String,
50}
51
52impl ListedGroupV1 {
53    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
54        Ok(Self {
55            group_id: decoder.read_string()?,
56            protocol_type: decoder.read_string()?,
57        })
58    }
59}
60
61#[cfg(test)]
62#[allow(clippy::unwrap_used)]
63mod tests {
64    use super::{ListGroupsRequestV1, ListGroupsResponseV1, API_KEY};
65    use crate::codec::{Decoder, Encoder};
66
67    #[test]
68    fn encodes_list_groups_v1_request() {
69        let request = ListGroupsRequestV1 {
70            correlation_id: 17,
71            client_id: Some("kafrust".to_owned()),
72        };
73
74        assert_eq!(
75            request.encode().unwrap(),
76            [
77                0, 16, // API key
78                0, 1, // API version
79                0, 0, 0, 17, // correlation ID
80                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // client ID
81            ]
82        );
83        assert_eq!(API_KEY, 16);
84    }
85
86    #[test]
87    fn decodes_list_groups_v1_response() {
88        let mut bytes = Encoder::new();
89        bytes.write_i32(4);
90        bytes.write_i16(0);
91        bytes.write_i32(2);
92        bytes.write_string("orders").unwrap();
93        bytes.write_string("consumer").unwrap();
94        bytes.write_string("connect-cluster").unwrap();
95        bytes.write_string("connect").unwrap();
96        let bytes = bytes.into_bytes();
97        let mut decoder = Decoder::new(&bytes);
98
99        let response = ListGroupsResponseV1::decode_body(&mut decoder).unwrap();
100
101        assert_eq!(response.throttle_time_ms, 4);
102        assert_eq!(response.error_code, 0);
103        assert_eq!(response.groups.len(), 2);
104        assert_eq!(response.groups[0].group_id, "orders");
105        assert_eq!(response.groups[0].protocol_type, "consumer");
106        assert_eq!(response.groups[1].protocol_type, "connect");
107        assert!(decoder.is_empty());
108    }
109}
kafrust_protocol/api/list_groups.rs

kafrust_protocol/api/
list_groups.rs
