kafrust_protocol/api/
describe_configs.rs

1use crate::codec::{Decoder, Encoder};
2use crate::error::Result;
3use crate::header::RequestHeader;
4
5pub const API_KEY: i16 = 32;
6
7#[derive(Debug, Clone, PartialEq, Eq)]
8pub struct DescribeConfigsRequestV1 {
9    pub correlation_id: i32,
10    pub client_id: Option<String>,
11    pub resources: Vec<DescribeConfigsResourceV1>,
12    pub include_synonyms: bool,
13}
14
15impl DescribeConfigsRequestV1 {
16    pub fn encode(&self) -> Result<Vec<u8>> {
17        let mut encoder = Encoder::new();
18        RequestHeader {
19            api_key: API_KEY,
20            api_version: 1,
21            correlation_id: self.correlation_id,
22            client_id: self.client_id.clone(),
23        }
24        .encode_v1(&mut encoder)?;
25        encoder.write_array(Some(&self.resources), |encoder, resource| {
26            resource.encode(encoder)
27        })?;
28        encoder.write_bool(self.include_synonyms);
29        Ok(encoder.into_bytes())
30    }
31}
32
33#[derive(Debug, Clone, PartialEq, Eq)]
34pub struct DescribeConfigsResourceV1 {
35    pub resource_type: i8,
36    pub resource_name: String,
37    pub configuration_keys: Option<Vec<String>>,
38}
39
40impl DescribeConfigsResourceV1 {
41    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
42        encoder.write_i8(self.resource_type);
43        encoder.write_string(&self.resource_name)?;
44        encoder.write_array(
45            self.configuration_keys.as_deref(),
46            |encoder, configuration_key| encoder.write_string(configuration_key),
47        )
48    }
49}
50
51#[derive(Debug, Clone, PartialEq, Eq)]
52pub struct DescribeConfigsResponseV1 {
53    pub throttle_time_ms: i32,
54    pub results: Vec<DescribeConfigsResultV1>,
55}
56
57impl DescribeConfigsResponseV1 {
58    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
59        Ok(Self {
60            throttle_time_ms: decoder.read_i32()?,
61            results: decoder
62                .read_array("describe configs results", DescribeConfigsResultV1::decode)?
63                .unwrap_or_default(),
64        })
65    }
66}
67
68#[derive(Debug, Clone, PartialEq, Eq)]
69pub struct DescribeConfigsResultV1 {
70    pub error_code: i16,
71    pub error_message: Option<String>,
72    pub resource_type: i8,
73    pub resource_name: String,
74    pub configs: Vec<DescribeConfigsEntryV1>,
75}
76
77impl DescribeConfigsResultV1 {
78    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
79        Ok(Self {
80            error_code: decoder.read_i16()?,
81            error_message: decoder.read_nullable_string()?,
82            resource_type: decoder.read_i8()?,
83            resource_name: decoder.read_string()?,
84            configs: decoder
85                .read_array("describe configs entries", DescribeConfigsEntryV1::decode)?
86                .unwrap_or_default(),
87        })
88    }
89}
90
91#[derive(Debug, Clone, PartialEq, Eq)]
92pub struct DescribeConfigsEntryV1 {
93    pub name: String,
94    pub value: Option<String>,
95    pub read_only: bool,
96    pub config_source: i8,
97    pub is_sensitive: bool,
98    pub synonyms: Vec<DescribeConfigsSynonymV1>,
99}
100
101impl DescribeConfigsEntryV1 {
102    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
103        Ok(Self {
104            name: decoder.read_string()?,
105            value: decoder.read_nullable_string()?,
106            read_only: decoder.read_bool()?,
107            config_source: decoder.read_i8()?,
108            is_sensitive: decoder.read_bool()?,
109            synonyms: decoder
110                .read_array(
111                    "describe configs synonyms",
112                    DescribeConfigsSynonymV1::decode,
113                )?
114                .unwrap_or_default(),
115        })
116    }
117}
118
119#[derive(Debug, Clone, PartialEq, Eq)]
120pub struct DescribeConfigsSynonymV1 {
121    pub name: String,
122    pub value: Option<String>,
123    pub source: i8,
124}
125
126impl DescribeConfigsSynonymV1 {
127    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
128        Ok(Self {
129            name: decoder.read_string()?,
130            value: decoder.read_nullable_string()?,
131            source: decoder.read_i8()?,
132        })
133    }
134}
135
136#[cfg(test)]
137#[allow(clippy::unwrap_used)]
138mod tests {
139    use super::{
140        DescribeConfigsRequestV1, DescribeConfigsResourceV1, DescribeConfigsResponseV1, API_KEY,
141    };
142    use crate::codec::Decoder;
143
144    #[test]
145    fn encodes_describe_configs_v1_request() {
146        let request = DescribeConfigsRequestV1 {
147            correlation_id: 12,
148            client_id: Some("kafrust".to_owned()),
149            resources: vec![DescribeConfigsResourceV1 {
150                resource_type: 2,
151                resource_name: "orders".to_owned(),
152                configuration_keys: Some(vec!["cleanup.policy".to_owned()]),
153            }],
154            include_synonyms: true,
155        };
156
157        assert_eq!(
158            request.encode().unwrap(),
159            [
160                0, 32, // API key
161                0, 1, // API version
162                0, 0, 0, 12, // correlation ID
163                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // client ID
164                0, 0, 0, 1, // resource count
165                2, // topic resource
166                0, 6, b'o', b'r', b'd', b'e', b'r', b's', // resource name
167                0, 0, 0, 1, // configuration key count
168                0, 14, b'c', b'l', b'e', b'a', b'n', b'u', b'p', b'.', b'p', b'o', b'l', b'i',
169                b'c', b'y', // configuration key
170                1,    // include synonyms
171            ]
172        );
173        assert_eq!(API_KEY, 32);
174    }
175
176    #[test]
177    fn decodes_describe_configs_v1_response() {
178        let bytes = [
179            0, 0, 0, 9, // throttle time
180            0, 0, 0, 1, // result count
181            0, 0, // success
182            0xff, 0xff, // null error message
183            2,    // topic resource
184            0, 6, b'o', b'r', b'd', b'e', b'r', b's', // resource name
185            0, 0, 0, 1, // config count
186            0, 14, b'c', b'l', b'e', b'a', b'n', b'u', b'p', b'.', b'p', b'o', b'l', b'i', b'c',
187            b'y', // config name
188            0, 7, b'c', b'o', b'm', b'p', b'a', b'c', b't', // value
189            0,    // read only
190            1,    // dynamic topic config source
191            0,    // not sensitive
192            0, 0, 0, 1, // synonym count
193            0, 14, b'c', b'l', b'e', b'a', b'n', b'u', b'p', b'.', b'p', b'o', b'l', b'i', b'c',
194            b'y', // synonym name
195            0, 6, b'd', b'e', b'l', b'e', b't', b'e', // synonym value
196            5,    // default config source
197        ];
198        let mut decoder = Decoder::new(&bytes);
199
200        let response = DescribeConfigsResponseV1::decode_body(&mut decoder).unwrap();
201
202        assert_eq!(response.throttle_time_ms, 9);
203        assert_eq!(response.results.len(), 1);
204        let result = &response.results[0];
205        assert_eq!(result.resource_type, 2);
206        assert_eq!(result.resource_name, "orders");
207        assert_eq!(result.configs.len(), 1);
208        assert_eq!(result.configs[0].name, "cleanup.policy");
209        assert_eq!(result.configs[0].value.as_deref(), Some("compact"));
210        assert_eq!(result.configs[0].config_source, 1);
211        assert_eq!(result.configs[0].synonyms[0].source, 5);
212        assert!(decoder.is_empty());
213    }
214}
kafrust_protocol/api/describe_configs.rs

kafrust_protocol/api/
describe_configs.rs
