kafrust_protocol/api/
heartbeat.rs

1use crate::codec::{Decoder, Encoder};
2use crate::error::Result;
3use crate::header::RequestHeader;
4
5pub const API_KEY: i16 = 12;
6
7#[derive(Debug, Clone, PartialEq, Eq)]
8pub struct HeartbeatRequestV2 {
9    pub correlation_id: i32,
10    pub client_id: Option<String>,
11    pub group_id: String,
12    pub generation_id: i32,
13    pub member_id: String,
14}
15
16impl HeartbeatRequestV2 {
17    pub fn encode(&self) -> Result<Vec<u8>> {
18        let mut encoder = Encoder::new();
19        RequestHeader {
20            api_key: API_KEY,
21            api_version: 2,
22            correlation_id: self.correlation_id,
23            client_id: self.client_id.clone(),
24        }
25        .encode_v1(&mut encoder)?;
26        encoder.write_string(&self.group_id)?;
27        encoder.write_i32(self.generation_id);
28        encoder.write_string(&self.member_id)?;
29        Ok(encoder.into_bytes())
30    }
31}
32
33#[derive(Debug, Clone, PartialEq, Eq)]
34pub struct HeartbeatResponseV2 {
35    pub throttle_time_ms: i32,
36    pub error_code: i16,
37}
38
39impl HeartbeatResponseV2 {
40    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
41        Ok(Self {
42            throttle_time_ms: decoder.read_i32()?,
43            error_code: decoder.read_i16()?,
44        })
45    }
46}
47
48#[cfg(test)]
49#[allow(clippy::unwrap_used)]
50mod tests {
51    use super::{HeartbeatRequestV2, HeartbeatResponseV2};
52    use crate::codec::Decoder;
53
54    #[test]
55    fn encodes_heartbeat_v2_request() {
56        let request = HeartbeatRequestV2 {
57            correlation_id: 17,
58            client_id: Some("kafrust".to_owned()),
59            group_id: "orders-group".to_owned(),
60            generation_id: 7,
61            member_id: "member-a".to_owned(),
62        };
63
64        assert_eq!(
65            request.encode().unwrap(),
66            [
67                0, 12, // api key
68                0, 2, // api version
69                0, 0, 0, 17, // correlation id
70                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // client id
71                0, 12, b'o', b'r', b'd', b'e', b'r', b's', b'-', b'g', b'r', b'o', b'u',
72                b'p', // group id
73                0, 0, 0, 7, // generation id
74                0, 8, b'm', b'e', b'm', b'b', b'e', b'r', b'-', b'a', // member id
75            ]
76        );
77    }
78
79    #[test]
80    fn decodes_heartbeat_v2_response() {
81        let bytes = [
82            0, 0, 0, 0, // throttle time
83            0, 0, // error code
84        ];
85        let mut decoder = Decoder::new(&bytes);
86        let response = HeartbeatResponseV2::decode_body(&mut decoder).unwrap();
87
88        assert_eq!(response.throttle_time_ms, 0);
89        assert_eq!(response.error_code, 0);
90        assert!(decoder.is_empty());
91    }
92}
kafrust_protocol/api/heartbeat.rs

kafrust_protocol/api/
heartbeat.rs
