kafrust_protocol/api/
api_versions.rs

1use crate::codec::{Decoder, Encoder};
2use crate::error::Result;
3use crate::header::RequestHeader;
4
5pub const API_KEY: i16 = 18;
6
7#[derive(Debug, Clone, Copy, PartialEq, Eq)]
8pub struct ApiKeyVersion {
9    pub api_key: i16,
10    pub min_version: i16,
11    pub max_version: i16,
12}
13
14impl ApiKeyVersion {
15    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
16        Ok(Self {
17            api_key: decoder.read_i16()?,
18            min_version: decoder.read_i16()?,
19            max_version: decoder.read_i16()?,
20        })
21    }
22}
23
24#[derive(Debug, Clone, PartialEq, Eq)]
25pub struct ApiVersionsRequestV0 {
26    pub correlation_id: i32,
27    pub client_id: Option<String>,
28}
29
30impl ApiVersionsRequestV0 {
31    pub fn encode(&self) -> Result<Vec<u8>> {
32        let mut encoder = Encoder::new();
33        RequestHeader {
34            api_key: API_KEY,
35            api_version: 0,
36            correlation_id: self.correlation_id,
37            client_id: self.client_id.clone(),
38        }
39        .encode_v1(&mut encoder)?;
40        Ok(encoder.into_bytes())
41    }
42}
43
44#[derive(Debug, Clone, PartialEq, Eq)]
45pub struct ApiVersionsResponseV0 {
46    pub error_code: i16,
47    pub api_keys: Vec<ApiKeyVersion>,
48}
49
50impl ApiVersionsResponseV0 {
51    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
52        let error_code = decoder.read_i16()?;
53        let api_keys = decoder
54            .read_array("api versions", ApiKeyVersion::decode)?
55            .unwrap_or_default();
56        Ok(Self {
57            error_code,
58            api_keys,
59        })
60    }
61
62    pub fn highest_supported_version(&self, api_key: i16, max_supported: i16) -> Option<i16> {
63        self.api_keys
64            .iter()
65            .find(|version| version.api_key == api_key)
66            .and_then(|version| {
67                let selected = version.max_version.min(max_supported);
68                (selected >= version.min_version).then_some(selected)
69            })
70    }
71}
72
73#[cfg(test)]
74#[allow(clippy::unwrap_used)]
75mod tests {
76    use super::{ApiVersionsRequestV0, ApiVersionsResponseV0, API_KEY};
77    use crate::codec::Decoder;
78
79    #[test]
80    fn encodes_api_versions_request_v0() {
81        let request = ApiVersionsRequestV0 {
82            correlation_id: 42,
83            client_id: Some("kafrust".to_owned()),
84        };
85        assert_eq!(
86            request.encode().unwrap(),
87            [
88                0, 18, // api key
89                0, 0, // api version
90                0, 0, 0, 42, // correlation id
91                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't',
92            ]
93        );
94        assert_eq!(API_KEY, 18);
95    }
96
97    #[test]
98    fn decodes_api_versions_response_v0() {
99        let bytes = [
100            0, 0, // error code
101            0, 0, 0, 2, // api key count
102            0, 18, 0, 0, 0, 4, // ApiVersions min/max
103            0, 3, 0, 1, 0, 9, // Metadata min/max
104        ];
105        let mut decoder = Decoder::new(&bytes);
106        let response = ApiVersionsResponseV0::decode_body(&mut decoder).unwrap();
107
108        assert_eq!(response.error_code, 0);
109        assert_eq!(response.api_keys.len(), 2);
110        assert_eq!(response.highest_supported_version(18, 3), Some(3));
111        assert_eq!(response.highest_supported_version(3, 12), Some(9));
112        assert_eq!(response.highest_supported_version(1, 1), None);
113        assert!(decoder.is_empty());
114    }
115}
kafrust_protocol/api/api_versions.rs

kafrust_protocol/api/
api_versions.rs
