kafrust_protocol/api/
alter_client_quotas.rs

1use crate::codec::{Decoder, Encoder};
2use crate::error::Result;
3use crate::header::RequestHeader;
4
5pub const API_KEY: i16 = 49;
6
7#[derive(Debug, Clone, PartialEq)]
8pub struct AlterClientQuotasRequestV0 {
9    pub correlation_id: i32,
10    pub client_id: Option<String>,
11    pub entries: Vec<AlterClientQuotasEntryV0>,
12    pub validate_only: bool,
13}
14
15impl AlterClientQuotasRequestV0 {
16    pub fn encode(&self) -> Result<Vec<u8>> {
17        let mut encoder = Encoder::new();
18        RequestHeader {
19            api_key: API_KEY,
20            api_version: 0,
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.entries), |encoder, entry| {
26            encoder.write_array(Some(&entry.entities), |encoder, entity| {
27                encoder.write_string(&entity.entity_type)?;
28                encoder.write_nullable_string(entity.entity_name.as_deref())?;
29                Ok(())
30            })?;
31            encoder.write_array(Some(&entry.operations), |encoder, operation| {
32                encoder.write_string(&operation.key)?;
33                encoder.write_f64(operation.value);
34                encoder.write_bool(operation.remove);
35                Ok(())
36            })?;
37            Ok(())
38        })?;
39        encoder.write_bool(self.validate_only);
40        Ok(encoder.into_bytes())
41    }
42}
43
44#[derive(Debug, Clone, PartialEq)]
45pub struct AlterClientQuotasEntryV0 {
46    pub entities: Vec<AlterClientQuotasEntityV0>,
47    pub operations: Vec<AlterClientQuotasOperationV0>,
48}
49
50#[derive(Debug, Clone, PartialEq, Eq)]
51pub struct AlterClientQuotasEntityV0 {
52    pub entity_type: String,
53    pub entity_name: Option<String>,
54}
55
56#[derive(Debug, Clone, PartialEq)]
57pub struct AlterClientQuotasOperationV0 {
58    pub key: String,
59    pub value: f64,
60    pub remove: bool,
61}
62
63#[derive(Debug, Clone, PartialEq, Eq)]
64pub struct AlterClientQuotasResponseV0 {
65    pub throttle_time_ms: i32,
66    pub entries: Vec<AlterClientQuotasResultV0>,
67}
68
69impl AlterClientQuotasResponseV0 {
70    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
71        Ok(Self {
72            throttle_time_ms: decoder.read_i32()?,
73            entries: decoder
74                .read_array(
75                    "alter client quota results",
76                    AlterClientQuotasResultV0::decode,
77                )?
78                .unwrap_or_default(),
79        })
80    }
81}
82
83#[derive(Debug, Clone, PartialEq, Eq)]
84pub struct AlterClientQuotasResultV0 {
85    pub error_code: i16,
86    pub error_message: Option<String>,
87    pub entities: Vec<AlterClientQuotasEntityV0>,
88}
89
90impl AlterClientQuotasResultV0 {
91    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
92        Ok(Self {
93            error_code: decoder.read_i16()?,
94            error_message: decoder.read_nullable_string()?,
95            entities: decoder
96                .read_array(
97                    "alter client quota result entities",
98                    AlterClientQuotasEntityV0::decode,
99                )?
100                .unwrap_or_default(),
101        })
102    }
103}
104
105impl AlterClientQuotasEntityV0 {
106    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
107        Ok(Self {
108            entity_type: decoder.read_string()?,
109            entity_name: decoder.read_nullable_string()?,
110        })
111    }
112}
113
114#[cfg(test)]
115#[allow(clippy::unwrap_used)]
116mod tests {
117    use super::{
118        AlterClientQuotasEntityV0, AlterClientQuotasEntryV0, AlterClientQuotasOperationV0,
119        AlterClientQuotasRequestV0, AlterClientQuotasResponseV0, API_KEY,
120    };
121    use crate::codec::{Decoder, Encoder};
122
123    #[test]
124    fn encodes_alter_client_quotas_v0_request() {
125        let request = AlterClientQuotasRequestV0 {
126            correlation_id: 19,
127            client_id: None,
128            entries: vec![AlterClientQuotasEntryV0 {
129                entities: vec![AlterClientQuotasEntityV0 {
130                    entity_type: "user".to_owned(),
131                    entity_name: Some("alice".to_owned()),
132                }],
133                operations: vec![AlterClientQuotasOperationV0 {
134                    key: "producer_byte_rate".to_owned(),
135                    value: 1024.5,
136                    remove: false,
137                }],
138            }],
139            validate_only: true,
140        };
141
142        let bytes = request.encode().unwrap();
143        assert_eq!(&bytes[0..4], &[0, API_KEY as u8, 0, 0]);
144        assert_eq!(&bytes[4..8], &[0, 0, 0, 19]);
145        assert_eq!(bytes.last(), Some(&1));
146    }
147
148    #[test]
149    fn decodes_alter_client_quotas_v0_response() {
150        let mut bytes = Encoder::new();
151        bytes.write_i32(5);
152        bytes.write_i32(1);
153        bytes.write_i16(0);
154        bytes.write_nullable_string(None).unwrap();
155        bytes.write_i32(1);
156        bytes.write_string("user").unwrap();
157        bytes.write_nullable_string(Some("alice")).unwrap();
158        let bytes = bytes.into_bytes();
159        let mut decoder = Decoder::new(&bytes);
160
161        let response = AlterClientQuotasResponseV0::decode_body(&mut decoder).unwrap();
162
163        assert_eq!(response.throttle_time_ms, 5);
164        assert_eq!(response.entries[0].entities[0].entity_type, "user");
165        assert_eq!(
166            response.entries[0].entities[0].entity_name.as_deref(),
167            Some("alice")
168        );
169        assert!(decoder.is_empty());
170    }
171}
kafrust_protocol/api/alter_client_quotas.rs

kafrust_protocol/api/
alter_client_quotas.rs
