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/// Kafka ListGroups v4 request.
28///
29/// Version 4 adds a broker-side group-state filter and uses the flexible
30/// request header/body encoding.
31#[derive(Debug, Clone, PartialEq, Eq)]
32pub struct ListGroupsRequestV4 {
33    pub correlation_id: i32,
34    pub client_id: Option<String>,
35    pub states_filter: Vec<String>,
36}
37
38impl ListGroupsRequestV4 {
39    pub fn encode(&self) -> Result<Vec<u8>> {
40        encode_flexible_request(
41            4,
42            self.correlation_id,
43            self.client_id.as_deref(),
44            &self.states_filter,
45            &[],
46        )
47    }
48}
49
50/// Kafka ListGroups v5 request.
51///
52/// Version 5 adds a group-type filter to the v4 state filter.
53#[derive(Debug, Clone, PartialEq, Eq)]
54pub struct ListGroupsRequestV5 {
55    pub correlation_id: i32,
56    pub client_id: Option<String>,
57    pub states_filter: Vec<String>,
58    pub types_filter: Vec<String>,
59}
60
61impl ListGroupsRequestV5 {
62    pub fn encode(&self) -> Result<Vec<u8>> {
63        encode_flexible_request(
64            5,
65            self.correlation_id,
66            self.client_id.as_deref(),
67            &self.states_filter,
68            &self.types_filter,
69        )
70    }
71}
72
73fn encode_flexible_request(
74    api_version: i16,
75    correlation_id: i32,
76    client_id: Option<&str>,
77    states_filter: &[String],
78    types_filter: &[String],
79) -> Result<Vec<u8>> {
80    let mut encoder = Encoder::new();
81    RequestHeader {
82        api_key: API_KEY,
83        api_version,
84        correlation_id,
85        client_id: client_id.map(str::to_owned),
86    }
87    .encode_v2(&mut encoder)?;
88    encoder.write_compact_array(Some(states_filter), |encoder, state| {
89        encoder.write_compact_string(state)
90    })?;
91    if api_version >= 5 {
92        encoder.write_compact_array(Some(types_filter), |encoder, group_type| {
93            encoder.write_compact_string(group_type)
94        })?;
95    }
96    encoder.write_empty_tagged_fields();
97    Ok(encoder.into_bytes())
98}
99
100#[derive(Debug, Clone, PartialEq, Eq)]
101pub struct ListGroupsResponseV1 {
102    pub throttle_time_ms: i32,
103    pub error_code: i16,
104    pub groups: Vec<ListedGroupV1>,
105}
106
107impl ListGroupsResponseV1 {
108    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
109        Ok(Self {
110            throttle_time_ms: decoder.read_i32()?,
111            error_code: decoder.read_i16()?,
112            groups: decoder
113                .read_array("listed groups", ListedGroupV1::decode)?
114                .unwrap_or_default(),
115        })
116    }
117}
118
119/// Kafka ListGroups v4 response.
120#[derive(Debug, Clone, PartialEq, Eq)]
121pub struct ListGroupsResponseV4 {
122    pub throttle_time_ms: i32,
123    pub error_code: i16,
124    pub groups: Vec<ListedGroupV4>,
125}
126
127impl ListGroupsResponseV4 {
128    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
129        let throttle_time_ms = decoder.read_i32()?;
130        let error_code = decoder.read_i16()?;
131        let groups = decoder
132            .read_compact_array("listed groups", ListedGroupV4::decode)?
133            .unwrap_or_default();
134        decoder.read_tagged_fields()?;
135        Ok(Self {
136            throttle_time_ms,
137            error_code,
138            groups,
139        })
140    }
141}
142
143/// Kafka ListGroups v5 response.
144#[derive(Debug, Clone, PartialEq, Eq)]
145pub struct ListGroupsResponseV5 {
146    pub throttle_time_ms: i32,
147    pub error_code: i16,
148    pub groups: Vec<ListedGroupV5>,
149}
150
151impl ListGroupsResponseV5 {
152    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
153        let throttle_time_ms = decoder.read_i32()?;
154        let error_code = decoder.read_i16()?;
155        let groups = decoder
156            .read_compact_array("listed groups", ListedGroupV5::decode)?
157            .unwrap_or_default();
158        decoder.read_tagged_fields()?;
159        Ok(Self {
160            throttle_time_ms,
161            error_code,
162            groups,
163        })
164    }
165}
166
167#[derive(Debug, Clone, PartialEq, Eq)]
168pub struct ListedGroupV1 {
169    pub group_id: String,
170    pub protocol_type: String,
171}
172
173impl ListedGroupV1 {
174    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
175        Ok(Self {
176            group_id: decoder.read_string()?,
177            protocol_type: decoder.read_string()?,
178        })
179    }
180}
181
182/// One group returned by ListGroups v4.
183#[derive(Debug, Clone, PartialEq, Eq)]
184pub struct ListedGroupV4 {
185    pub group_id: String,
186    pub protocol_type: String,
187    pub group_state: String,
188}
189
190impl ListedGroupV4 {
191    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
192        let group = Self {
193            group_id: decoder.read_compact_string()?,
194            protocol_type: decoder.read_compact_string()?,
195            group_state: decoder.read_compact_string()?,
196        };
197        decoder.read_tagged_fields()?;
198        Ok(group)
199    }
200}
201
202/// One group returned by ListGroups v5.
203#[derive(Debug, Clone, PartialEq, Eq)]
204pub struct ListedGroupV5 {
205    pub group_id: String,
206    pub protocol_type: String,
207    pub group_state: String,
208    pub group_type: String,
209}
210
211impl ListedGroupV5 {
212    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
213        let group = Self {
214            group_id: decoder.read_compact_string()?,
215            protocol_type: decoder.read_compact_string()?,
216            group_state: decoder.read_compact_string()?,
217            group_type: decoder.read_compact_string()?,
218        };
219        decoder.read_tagged_fields()?;
220        Ok(group)
221    }
222}
223
224#[cfg(test)]
225#[allow(clippy::unwrap_used)]
226mod tests {
227    use super::{
228        ListGroupsRequestV1, ListGroupsRequestV4, ListGroupsRequestV5, ListGroupsResponseV1,
229        ListGroupsResponseV4, ListGroupsResponseV5, API_KEY,
230    };
231    use crate::codec::{Decoder, Encoder};
232
233    #[test]
234    fn encodes_list_groups_v1_request() {
235        let request = ListGroupsRequestV1 {
236            correlation_id: 17,
237            client_id: Some("kafrust".to_owned()),
238        };
239
240        assert_eq!(
241            request.encode().unwrap(),
242            [
243                0, 16, // API key
244                0, 1, // API version
245                0, 0, 0, 17, // correlation ID
246                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // client ID
247            ]
248        );
249        assert_eq!(API_KEY, 16);
250    }
251
252    #[test]
253    fn decodes_list_groups_v1_response() {
254        let mut bytes = Encoder::new();
255        bytes.write_i32(4);
256        bytes.write_i16(0);
257        bytes.write_i32(2);
258        bytes.write_string("orders").unwrap();
259        bytes.write_string("consumer").unwrap();
260        bytes.write_string("connect-cluster").unwrap();
261        bytes.write_string("connect").unwrap();
262        let bytes = bytes.into_bytes();
263        let mut decoder = Decoder::new(&bytes);
264
265        let response = ListGroupsResponseV1::decode_body(&mut decoder).unwrap();
266
267        assert_eq!(response.throttle_time_ms, 4);
268        assert_eq!(response.error_code, 0);
269        assert_eq!(response.groups.len(), 2);
270        assert_eq!(response.groups[0].group_id, "orders");
271        assert_eq!(response.groups[0].protocol_type, "consumer");
272        assert_eq!(response.groups[1].protocol_type, "connect");
273        assert!(decoder.is_empty());
274    }
275
276    #[test]
277    fn encodes_list_groups_v4_request_with_state_filter() {
278        let request = ListGroupsRequestV4 {
279            correlation_id: 17,
280            client_id: Some("kafrust".to_owned()),
281            states_filter: vec!["Stable".to_owned(), "Empty".to_owned()],
282        };
283
284        assert_eq!(
285            request.encode().unwrap(),
286            [
287                0, 16, // API key
288                0, 4, // API version
289                0, 0, 0, 17, // correlation ID
290                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // client ID
291                0,    // request header tagged fields
292                3,    // compact array length + 1
293                7, b'S', b't', b'a', b'b', b'l', b'e', 6, b'E', b'm', b'p', b't', b'y',
294                0, // request tagged fields
295            ]
296        );
297    }
298
299    #[test]
300    fn encodes_list_groups_v5_request_with_state_and_type_filters() {
301        let request = ListGroupsRequestV5 {
302            correlation_id: 17,
303            client_id: None,
304            states_filter: vec!["Stable".to_owned()],
305            types_filter: vec!["consumer".to_owned()],
306        };
307
308        assert_eq!(
309            request.encode().unwrap(),
310            [
311                0, 16, // API key
312                0, 5, // API version
313                0, 0, 0, 17, // correlation ID
314                255, 255, // nullable client ID
315                0,   // request header tagged fields
316                2,   // states compact array length + 1
317                7, b'S', b't', b'a', b'b', b'l', b'e', 2, // types compact array length + 1
318                9, b'c', b'o', b'n', b's', b'u', b'm', b'e', b'r', 0, // request tagged fields
319            ]
320        );
321    }
322
323    #[test]
324    fn decodes_list_groups_v4_response() {
325        let mut bytes = Encoder::new();
326        bytes.write_i32(4);
327        bytes.write_i16(0);
328        bytes
329            .write_compact_array(Some(&[()]), |encoder, ()| {
330                encoder.write_compact_string("orders")?;
331                encoder.write_compact_string("consumer")?;
332                encoder.write_compact_string("Stable")?;
333                encoder.write_empty_tagged_fields();
334                Ok(())
335            })
336            .unwrap();
337        bytes.write_empty_tagged_fields();
338        let bytes = bytes.into_bytes();
339        let mut decoder = Decoder::new(&bytes);
340
341        let response = ListGroupsResponseV4::decode_body(&mut decoder).unwrap();
342
343        assert_eq!(response.groups[0].group_state, "Stable");
344        assert!(decoder.is_empty());
345    }
346
347    #[test]
348    fn decodes_list_groups_v5_response() {
349        let mut bytes = Encoder::new();
350        bytes.write_i32(4);
351        bytes.write_i16(0);
352        bytes
353            .write_compact_array(Some(&[()]), |encoder, ()| {
354                encoder.write_compact_string("orders")?;
355                encoder.write_compact_string("consumer")?;
356                encoder.write_compact_string("Stable")?;
357                encoder.write_compact_string("consumer")?;
358                encoder.write_empty_tagged_fields();
359                Ok(())
360            })
361            .unwrap();
362        bytes.write_empty_tagged_fields();
363        let bytes = bytes.into_bytes();
364        let mut decoder = Decoder::new(&bytes);
365
366        let response = ListGroupsResponseV5::decode_body(&mut decoder).unwrap();
367
368        assert_eq!(response.groups[0].group_type, "consumer");
369        assert!(decoder.is_empty());
370    }
371}
kafrust_protocol/api/list_groups.rs

kafrust_protocol/api/
list_groups.rs
