kafrust_protocol/
header.rs

1use crate::codec::{Decoder, Encoder};
2use crate::error::Result;
3
4#[derive(Debug, Clone, PartialEq, Eq)]
5pub struct RequestHeader {
6    pub api_key: i16,
7    pub api_version: i16,
8    pub correlation_id: i32,
9    pub client_id: Option<String>,
10}
11
12impl RequestHeader {
13    pub fn encode_v1(&self, encoder: &mut Encoder) -> Result<()> {
14        encoder.write_i16(self.api_key);
15        encoder.write_i16(self.api_version);
16        encoder.write_i32(self.correlation_id);
17        encoder.write_nullable_string(self.client_id.as_deref())
18    }
19
20    pub fn encode_v2(&self, encoder: &mut Encoder) -> Result<()> {
21        self.encode_v1(encoder)?;
22        encoder.write_empty_tagged_fields();
23        Ok(())
24    }
25}
26
27#[derive(Debug, Clone, PartialEq, Eq)]
28pub struct ResponseHeader {
29    pub correlation_id: i32,
30}
31
32impl ResponseHeader {
33    pub fn decode_v0(decoder: &mut Decoder<'_>) -> Result<Self> {
34        Ok(Self {
35            correlation_id: decoder.read_i32()?,
36        })
37    }
38
39    pub fn decode_v1(decoder: &mut Decoder<'_>) -> Result<Self> {
40        let header = Self::decode_v0(decoder)?;
41        let _tagged_fields = decoder.read_tagged_fields()?;
42        Ok(header)
43    }
44}
45
46#[cfg(test)]
47#[allow(clippy::unwrap_used)]
48mod tests {
49    use super::{RequestHeader, ResponseHeader};
50    use crate::codec::{Decoder, Encoder};
51
52    #[test]
53    fn encodes_request_header_v1() {
54        let header = RequestHeader {
55            api_key: 18,
56            api_version: 0,
57            correlation_id: 7,
58            client_id: Some("kafrust".to_owned()),
59        };
60        let mut encoder = Encoder::new();
61        header.encode_v1(&mut encoder).unwrap();
62        assert_eq!(
63            encoder.into_bytes(),
64            [0, 18, 0, 0, 0, 0, 0, 7, 0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't']
65        );
66    }
67
68    #[test]
69    fn encodes_request_header_v2_with_nullable_client_id_and_tagged_fields() {
70        let header = RequestHeader {
71            api_key: 3,
72            api_version: 12,
73            correlation_id: 7,
74            client_id: Some("kafrust".to_owned()),
75        };
76        let mut encoder = Encoder::new();
77        header.encode_v2(&mut encoder).unwrap();
78        assert_eq!(
79            encoder.into_bytes(),
80            [0, 3, 0, 12, 0, 0, 0, 7, 0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', 0]
81        );
82    }
83
84    #[test]
85    fn decodes_response_header_v0() {
86        let mut decoder = Decoder::new(&[0, 0, 0, 7]);
87        let header = ResponseHeader::decode_v0(&mut decoder).unwrap();
88        assert_eq!(header.correlation_id, 7);
89    }
90}
kafrust_protocol/header.rs

kafrust_protocol/
header.rs
