kafrust_protocol/api/
init_producer_id.rs

1use crate::codec::{Decoder, Encoder};
2use crate::error::Result;
3use crate::header::RequestHeader;
4
5pub const API_KEY: i16 = 22;
6
7#[derive(Debug, Clone, PartialEq, Eq)]
8pub struct InitProducerIdRequestV0 {
9    pub correlation_id: i32,
10    pub client_id: Option<String>,
11    pub transactional_id: Option<String>,
12    pub transaction_timeout_ms: i32,
13}
14
15impl InitProducerIdRequestV0 {
16    pub fn encode(&self) -> Result<Vec<u8>> {
17        let mut encoder = Encoder::new();
18        RequestHeader {
19            api_key: API_KEY,
20            api_version: 0,
21            correlation_id: self.correlation_id,
22            client_id: self.client_id.clone(),
23        }
24        .encode_v1(&mut encoder)?;
25        encoder.write_nullable_string(self.transactional_id.as_deref())?;
26        encoder.write_i32(self.transaction_timeout_ms);
27        Ok(encoder.into_bytes())
28    }
29}
30
31#[derive(Debug, Clone, PartialEq, Eq)]
32pub struct InitProducerIdResponseV0 {
33    pub throttle_time_ms: i32,
34    pub error_code: i16,
35    pub producer_id: i64,
36    pub producer_epoch: i16,
37}
38
39impl InitProducerIdResponseV0 {
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            producer_id: decoder.read_i64()?,
45            producer_epoch: decoder.read_i16()?,
46        })
47    }
48}
49
50#[cfg(test)]
51#[allow(clippy::unwrap_used)]
52mod tests {
53    use super::{InitProducerIdRequestV0, InitProducerIdResponseV0, API_KEY};
54    use crate::codec::Decoder;
55
56    #[test]
57    fn encodes_non_transactional_init_producer_id_v0_request() {
58        let request = InitProducerIdRequestV0 {
59            correlation_id: 23,
60            client_id: Some("kafrust".to_owned()),
61            transactional_id: None,
62            transaction_timeout_ms: 60_000,
63        };
64
65        assert_eq!(
66            request.encode().unwrap(),
67            [
68                0, 22, // api key
69                0, 0, // api version
70                0, 0, 0, 23, // correlation id
71                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // client id
72                0xff, 0xff, // null transactional id
73                0, 0, 0xea, 0x60, // transaction timeout
74            ]
75        );
76        assert_eq!(API_KEY, 22);
77    }
78
79    #[test]
80    fn encodes_transactional_init_producer_id_v0_request() {
81        let request = InitProducerIdRequestV0 {
82            correlation_id: 24,
83            client_id: None,
84            transactional_id: Some("orders-tx".to_owned()),
85            transaction_timeout_ms: 30_000,
86        };
87
88        assert_eq!(
89            request.encode().unwrap(),
90            [
91                0, 22, // api key
92                0, 0, // api version
93                0, 0, 0, 24, // correlation id
94                0xff, 0xff, // null client id
95                0, 9, b'o', b'r', b'd', b'e', b'r', b's', b'-', b't', b'x', 0, 0, 0x75,
96                0x30, // transaction timeout
97            ]
98        );
99    }
100
101    #[test]
102    fn decodes_init_producer_id_v0_response() {
103        let bytes = [
104            0, 0, 0, 12, // throttle time
105            0, 0, // error code
106            0, 0, 0, 0, 0, 0, 0, 42, // producer id
107            0, 3, // producer epoch
108        ];
109        let mut decoder = Decoder::new(&bytes);
110        let response = InitProducerIdResponseV0::decode_body(&mut decoder).unwrap();
111
112        assert_eq!(response.throttle_time_ms, 12);
113        assert_eq!(response.error_code, 0);
114        assert_eq!(response.producer_id, 42);
115        assert_eq!(response.producer_epoch, 3);
116        assert!(decoder.is_empty());
117    }
118}
kafrust_protocol/api/init_producer_id.rs

kafrust_protocol/api/
init_producer_id.rs
