kafrust_protocol/api/
unregister_broker.rs

1use crate::codec::{Decoder, Encoder};
2use crate::error::Result;
3use crate::header::RequestHeader;
4
5/// Kafka UnregisterBroker API key.
6pub const API_KEY: i16 = 64;
7
8/// UnregisterBroker v0 request.
9#[derive(Debug, Clone, PartialEq, Eq)]
10pub struct UnregisterBrokerRequestV0 {
11    pub correlation_id: i32,
12    pub client_id: Option<String>,
13    pub broker_id: i32,
14}
15
16impl UnregisterBrokerRequestV0 {
17    /// Encodes the flexible v0 request header and body.
18    pub fn encode(&self) -> Result<Vec<u8>> {
19        let mut encoder = Encoder::new();
20        RequestHeader {
21            api_key: API_KEY,
22            api_version: 0,
23            correlation_id: self.correlation_id,
24            client_id: self.client_id.clone(),
25        }
26        .encode_v2(&mut encoder)?;
27        encoder.write_i32(self.broker_id);
28        encoder.write_empty_tagged_fields();
29        Ok(encoder.into_bytes())
30    }
31}
32
33/// UnregisterBroker v0 response.
34#[derive(Debug, Clone, PartialEq, Eq)]
35pub struct UnregisterBrokerResponseV0 {
36    pub throttle_time_ms: i32,
37    pub error_code: i16,
38    pub error_message: Option<String>,
39}
40
41impl UnregisterBrokerResponseV0 {
42    /// Decodes a flexible UnregisterBroker response body.
43    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
44        let response = Self {
45            throttle_time_ms: decoder.read_i32()?,
46            error_code: decoder.read_i16()?,
47            error_message: decoder.read_compact_nullable_string()?,
48        };
49        decoder.read_tagged_fields()?;
50        Ok(response)
51    }
52}
53
54#[cfg(test)]
55#[allow(clippy::unwrap_used)]
56mod tests {
57    use super::{UnregisterBrokerRequestV0, UnregisterBrokerResponseV0, API_KEY};
58    use crate::codec::{Decoder, Encoder};
59
60    #[test]
61    fn encodes_unregister_broker_v0_wire_shape() {
62        let request = UnregisterBrokerRequestV0 {
63            correlation_id: 7,
64            client_id: Some("kafrust".to_owned()),
65            broker_id: 4,
66        };
67        let bytes = request.encode().unwrap();
68        let mut decoder = Decoder::new(&bytes);
69        assert_eq!(decoder.read_i16().unwrap(), API_KEY);
70        assert_eq!(decoder.read_i16().unwrap(), 0);
71        assert_eq!(decoder.read_i32().unwrap(), 7);
72        assert_eq!(
73            decoder.read_nullable_string().unwrap().as_deref(),
74            Some("kafrust")
75        );
76        decoder.read_tagged_fields().unwrap();
77        assert_eq!(decoder.read_i32().unwrap(), 4);
78        decoder.read_tagged_fields().unwrap();
79        assert!(decoder.is_empty());
80    }
81
82    #[test]
83    fn decodes_unregister_broker_v0_response() {
84        let mut encoder = Encoder::new();
85        encoder.write_i32(12);
86        encoder.write_i16(0);
87        encoder.write_compact_nullable_string(Some("ok")).unwrap();
88        encoder.write_empty_tagged_fields();
89        let bytes = encoder.into_bytes();
90        let mut decoder = Decoder::new(&bytes);
91        let response = UnregisterBrokerResponseV0::decode_body(&mut decoder).unwrap();
92        assert_eq!(response.throttle_time_ms, 12);
93        assert_eq!(response.error_code, 0);
94        assert_eq!(response.error_message.as_deref(), Some("ok"));
95        assert!(decoder.is_empty());
96    }
97}
kafrust_protocol/api/unregister_broker.rs

kafrust_protocol/api/
unregister_broker.rs
