kafrust_protocol/api/
find_coordinator.rs

1use crate::codec::{Decoder, Encoder};
2use crate::error::Result;
3use crate::header::RequestHeader;
4
5pub const API_KEY: i16 = 10;
6
7#[derive(Debug, Clone, Copy, PartialEq, Eq)]
8pub enum CoordinatorType {
9    Group,
10    Transaction,
11}
12
13impl CoordinatorType {
14    fn as_i8(self) -> i8 {
15        match self {
16            Self::Group => 0,
17            Self::Transaction => 1,
18        }
19    }
20}
21
22#[derive(Debug, Clone, PartialEq, Eq)]
23pub struct FindCoordinatorRequestV1 {
24    pub correlation_id: i32,
25    pub client_id: Option<String>,
26    pub coordinator_key: String,
27    pub coordinator_type: CoordinatorType,
28}
29
30impl FindCoordinatorRequestV1 {
31    pub fn encode(&self) -> Result<Vec<u8>> {
32        let mut encoder = Encoder::new();
33        RequestHeader {
34            api_key: API_KEY,
35            api_version: 1,
36            correlation_id: self.correlation_id,
37            client_id: self.client_id.clone(),
38        }
39        .encode_v1(&mut encoder)?;
40        encoder.write_string(&self.coordinator_key)?;
41        encoder.write_i8(self.coordinator_type.as_i8());
42        Ok(encoder.into_bytes())
43    }
44}
45
46#[derive(Debug, Clone, PartialEq, Eq)]
47pub struct FindCoordinatorResponseV1 {
48    pub throttle_time_ms: i32,
49    pub error_code: i16,
50    pub error_message: Option<String>,
51    pub node_id: i32,
52    pub host: String,
53    pub port: i32,
54}
55
56impl FindCoordinatorResponseV1 {
57    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
58        Ok(Self {
59            throttle_time_ms: decoder.read_i32()?,
60            error_code: decoder.read_i16()?,
61            error_message: decoder.read_nullable_string()?,
62            node_id: decoder.read_i32()?,
63            host: decoder.read_string()?,
64            port: decoder.read_i32()?,
65        })
66    }
67}
68
69#[cfg(test)]
70#[allow(clippy::unwrap_used)]
71mod tests {
72    use super::{CoordinatorType, FindCoordinatorRequestV1, FindCoordinatorResponseV1, API_KEY};
73    use crate::codec::Decoder;
74
75    #[test]
76    fn encodes_find_coordinator_v1_for_group() {
77        let request = FindCoordinatorRequestV1 {
78            correlation_id: 11,
79            client_id: Some("kafrust".to_owned()),
80            coordinator_key: "orders-group".to_owned(),
81            coordinator_type: CoordinatorType::Group,
82        };
83
84        assert_eq!(
85            request.encode().unwrap(),
86            [
87                0, 10, // api key
88                0, 1, // api version
89                0, 0, 0, 11, // correlation id
90                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // client id
91                0, 12, b'o', b'r', b'd', b'e', b'r', b's', b'-', b'g', b'r', b'o', b'u',
92                b'p', // coordinator key
93                0,    // group coordinator type
94            ]
95        );
96        assert_eq!(API_KEY, 10);
97    }
98
99    #[test]
100    fn encodes_find_coordinator_v1_for_transaction() {
101        let request = FindCoordinatorRequestV1 {
102            correlation_id: 12,
103            client_id: None,
104            coordinator_key: "orders-tx".to_owned(),
105            coordinator_type: CoordinatorType::Transaction,
106        };
107
108        assert_eq!(request.encode().unwrap().last(), Some(&1));
109    }
110
111    #[test]
112    fn decodes_find_coordinator_v1_response() {
113        let bytes = [
114            0, 0, 0, 0, // throttle time
115            0, 0, // error code
116            0xff, 0xff, // null error message
117            0, 0, 0, 2, // node id
118            0, 9, b'l', b'o', b'c', b'a', b'l', b'h', b'o', b's', b't', // host
119            0, 0, 35, 132, // port 9092
120        ];
121        let mut decoder = Decoder::new(&bytes);
122        let response = FindCoordinatorResponseV1::decode_body(&mut decoder).unwrap();
123
124        assert_eq!(response.throttle_time_ms, 0);
125        assert_eq!(response.error_code, 0);
126        assert_eq!(response.error_message, None);
127        assert_eq!(response.node_id, 2);
128        assert_eq!(response.host, "localhost");
129        assert_eq!(response.port, 9092);
130        assert!(decoder.is_empty());
131    }
132}
kafrust_protocol/api/find_coordinator.rs

kafrust_protocol/api/
find_coordinator.rs
