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        Ok(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    }
83}
84
85#[derive(Debug, Clone, PartialEq, Eq)]
86pub struct ListOffsetsTopicResponseV1 {
87    pub name: String,
88    pub partitions: Vec<ListOffsetsPartitionResponseV1>,
89}
90
91#[derive(Debug, Clone, PartialEq, Eq)]
92pub struct ListOffsetsPartitionResponseV1 {
93    pub partition_index: i32,
94    pub error_code: i16,
95    pub timestamp: i64,
96    pub offset: i64,
97}
98
99#[cfg(test)]
100#[allow(clippy::unwrap_used)]
101mod tests {
102    use super::*;
103
104    #[test]
105    fn encodes_list_offsets_v1_request() {
106        let request = ListOffsetsRequestV1 {
107            correlation_id: 7,
108            client_id: None,
109            replica_id: -1,
110            topics: vec![ListOffsetsTopicV1 {
111                name: "x".to_owned(),
112                partitions: vec![ListOffsetsPartitionV1 {
113                    partition_index: 2,
114                    timestamp: LATEST_TIMESTAMP,
115                }],
116            }],
117        };
118
119        assert_eq!(
120            request.encode().unwrap(),
121            [
122                0, 2, 0, 1, 0, 0, 0, 7, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0, 0, 0, 1, 0, 1, b'x',
123                0, 0, 0, 1, 0, 0, 0, 2, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
124            ]
125        );
126    }
127
128    #[test]
129    fn decodes_list_offsets_v1_response() {
130        let mut encoder = Encoder::new();
131        encoder.write_i32(1);
132        encoder.write_string("x").unwrap();
133        encoder.write_i32(1);
134        encoder.write_i32(2);
135        encoder.write_i16(0);
136        encoder.write_i64(123);
137        encoder.write_i64(42);
138        let bytes = encoder.into_bytes();
139
140        let response = ListOffsetsResponseV1::decode_body(&mut Decoder::new(&bytes)).unwrap();
141        assert_eq!(response.topics[0].partitions[0].offset, 42);
142    }
143}
kafrust_protocol/api/list_offsets.rs

kafrust_protocol/api/
list_offsets.rs
