kafrust_protocol/api/
list_offsets.rs

1use crate::codec::{Decoder, Encoder};
2use crate::error::Result;
3use crate::header::RequestHeader;
4
5pub const API_KEY: i16 = 2;
6pub const EARLIEST_TIMESTAMP: i64 = -2;
7pub const LATEST_TIMESTAMP: i64 = -1;
8
9#[derive(Debug, Clone, PartialEq, Eq)]
10pub struct ListOffsetsRequestV1 {
11    pub correlation_id: i32,
12    pub client_id: Option<String>,
13    pub replica_id: i32,
14    pub topics: Vec<ListOffsetsTopicV1>,
15}
16
17impl ListOffsetsRequestV1 {
18    pub fn encode(&self) -> Result<Vec<u8>> {
19        let mut encoder = Encoder::new();
20        RequestHeader {
21            api_key: API_KEY,
22            api_version: 1,
23            correlation_id: self.correlation_id,
24            client_id: self.client_id.clone(),
25        }
26        .encode_v1(&mut encoder)?;
27        encoder.write_i32(self.replica_id);
28        encoder.write_array(Some(&self.topics), |encoder, topic| topic.encode(encoder))?;
29        Ok(encoder.into_bytes())
30    }
31}
32
33#[derive(Debug, Clone, PartialEq, Eq)]
34pub struct ListOffsetsTopicV1 {
35    pub name: String,
36    pub partitions: Vec<ListOffsetsPartitionV1>,
37}
38
39impl ListOffsetsTopicV1 {
40    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
41        encoder.write_string(&self.name)?;
42        encoder.write_array(Some(&self.partitions), |encoder, partition| {
43            encoder.write_i32(partition.partition_index);
44            encoder.write_i64(partition.timestamp);
45            Ok(())
46        })
47    }
48}
49
50#[derive(Debug, Clone, PartialEq, Eq)]
51pub struct ListOffsetsPartitionV1 {
52    pub partition_index: i32,
53    pub timestamp: i64,
54}
55
56#[derive(Debug, Clone, PartialEq, Eq)]
57pub struct ListOffsetsResponseV1 {
58    pub topics: Vec<ListOffsetsTopicResponseV1>,
59}
60
61impl ListOffsetsResponseV1 {
62    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
63        let response = Self {
64            topics: decoder
65                .read_array("list offsets topic responses", |decoder| {
66                    Ok(ListOffsetsTopicResponseV1 {
67                        name: decoder.read_string()?,
68                        partitions: decoder
69                            .read_array("list offsets partition responses", |decoder| {
70                                Ok(ListOffsetsPartitionResponseV1 {
71                                    partition_index: decoder.read_i32()?,
72                                    error_code: decoder.read_i16()?,
73                                    timestamp: decoder.read_i64()?,
74                                    offset: decoder.read_i64()?,
75                                })
76                            })?
77                            .unwrap_or_default(),
78                    })
79                })?
80                .unwrap_or_default(),
81        };
82        decoder.finish()?;
83        Ok(response)
84    }
85}
86
87#[derive(Debug, Clone, PartialEq, Eq)]
88pub struct ListOffsetsTopicResponseV1 {
89    pub name: String,
90    pub partitions: Vec<ListOffsetsPartitionResponseV1>,
91}
92
93#[derive(Debug, Clone, PartialEq, Eq)]
94pub struct ListOffsetsPartitionResponseV1 {
95    pub partition_index: i32,
96    pub error_code: i16,
97    pub timestamp: i64,
98    pub offset: i64,
99}
100
101#[cfg(test)]
102#[allow(clippy::unwrap_used)]
103mod tests {
104    use super::*;
105
106    #[test]
107    fn encodes_list_offsets_v1_request() {
108        let request = ListOffsetsRequestV1 {
109            correlation_id: 7,
110            client_id: None,
111            replica_id: -1,
112            topics: vec![ListOffsetsTopicV1 {
113                name: "x".to_owned(),
114                partitions: vec![ListOffsetsPartitionV1 {
115                    partition_index: 2,
116                    timestamp: LATEST_TIMESTAMP,
117                }],
118            }],
119        };
120
121        assert_eq!(
122            request.encode().unwrap(),
123            [
124                0, 2, 0, 1, 0, 0, 0, 7, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0, 0, 0, 1, 0, 1, b'x',
125                0, 0, 0, 1, 0, 0, 0, 2, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
126            ]
127        );
128    }
129
130    #[test]
131    fn decodes_list_offsets_v1_response() {
132        let mut encoder = Encoder::new();
133        encoder.write_i32(1);
134        encoder.write_string("x").unwrap();
135        encoder.write_i32(1);
136        encoder.write_i32(2);
137        encoder.write_i16(0);
138        encoder.write_i64(123);
139        encoder.write_i64(42);
140        let bytes = encoder.into_bytes();
141
142        let response = ListOffsetsResponseV1::decode_body(&mut Decoder::new(&bytes)).unwrap();
143        assert_eq!(response.topics[0].partitions[0].offset, 42);
144    }
145}
kafrust_protocol/api/list_offsets.rs

kafrust_protocol/api/
list_offsets.rs
