kafrust_protocol/codec/
mod.rs

1mod decode;
2mod encode;
3
4pub use decode::{Decoder, TaggedField};
5pub use encode::Encoder;
6
7#[cfg(test)]
8#[allow(clippy::unwrap_used)]
9mod tests {
10    use super::{Decoder, Encoder};
11    use crate::Error;
12
13    #[test]
14    fn encodes_and_decodes_fixed_width_primitives() {
15        let mut encoder = Encoder::new();
16        encoder.write_bool(true);
17        encoder.write_i16(0x1234);
18        encoder.write_i32(0x1234_5678);
19        encoder.write_i64(0x0102_0304_0506_0708);
20
21        let bytes = encoder.into_bytes();
22        assert_eq!(
23            bytes,
24            [
25                1, 0x12, 0x34, 0x12, 0x34, 0x56, 0x78, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07,
26                0x08,
27            ]
28        );
29
30        let mut decoder = Decoder::new(&bytes);
31        assert!(decoder.read_bool().unwrap());
32        assert_eq!(decoder.read_i16().unwrap(), 0x1234);
33        assert_eq!(decoder.read_i32().unwrap(), 0x1234_5678);
34        assert_eq!(decoder.read_i64().unwrap(), 0x0102_0304_0506_0708);
35        assert!(decoder.is_empty());
36    }
37
38    #[test]
39    fn encodes_and_decodes_nullable_strings_and_bytes() {
40        let mut encoder = Encoder::new();
41        encoder.write_nullable_string(Some("topic")).unwrap();
42        encoder.write_nullable_string(None).unwrap();
43        encoder.write_nullable_bytes(Some(&[1, 2, 3])).unwrap();
44        encoder.write_nullable_bytes(None).unwrap();
45
46        let bytes = encoder.into_bytes();
47        let mut decoder = Decoder::new(&bytes);
48        assert_eq!(
49            decoder.read_nullable_string().unwrap(),
50            Some("topic".to_owned())
51        );
52        assert_eq!(decoder.read_nullable_string().unwrap(), None);
53        assert_eq!(decoder.read_nullable_bytes().unwrap(), Some(vec![1, 2, 3]));
54        assert_eq!(decoder.read_nullable_bytes().unwrap(), None);
55        assert!(decoder.is_empty());
56    }
57
58    #[test]
59    fn encodes_and_decodes_compact_types_and_empty_tags() {
60        let mut encoder = Encoder::new();
61        encoder.write_unsigned_varint(300);
62        encoder.write_compact_string("kafka").unwrap();
63        encoder.write_compact_nullable_string(None).unwrap();
64        encoder.write_compact_bytes(&[9, 8]).unwrap();
65        encoder.write_empty_tagged_fields();
66
67        let bytes = encoder.into_bytes();
68        let mut decoder = Decoder::new(&bytes);
69        assert_eq!(decoder.read_unsigned_varint().unwrap(), 300);
70        assert_eq!(decoder.read_compact_string().unwrap(), "kafka");
71        assert_eq!(decoder.read_compact_nullable_string().unwrap(), None);
72        assert_eq!(decoder.read_compact_bytes().unwrap(), vec![9, 8]);
73        assert_eq!(decoder.read_tagged_fields().unwrap(), Vec::new());
74        assert!(decoder.is_empty());
75    }
76
77    #[test]
78    fn decodes_signed_varints() {
79        let mut decoder = Decoder::new(&[0x02, 0x01, 0x04, 0x03]);
80
81        assert_eq!(decoder.read_varint().unwrap(), 1);
82        assert_eq!(decoder.read_varint().unwrap(), -1);
83        assert_eq!(decoder.read_varlong().unwrap(), 2);
84        assert_eq!(decoder.read_varlong().unwrap(), -2);
85        assert!(decoder.is_empty());
86    }
87
88    #[test]
89    fn rejects_invalid_bool_and_short_input() {
90        let mut decoder = Decoder::new(&[2]);
91        assert_eq!(decoder.read_bool(), Err(Error::InvalidBool(2)));
92
93        let mut decoder = Decoder::new(&[0]);
94        assert!(matches!(
95            decoder.read_i16(),
96            Err(Error::UnexpectedEof {
97                needed: 2,
98                remaining: 1
99            })
100        ));
101    }
102}
kafrust_protocol/codec/mod.rs

kafrust_protocol/codec/
mod.rs
