krabka_records_legacy/message/
encode.rs

1//! Encoding one v0/v1 message: the byte count it occupies inside its frame,
2//! and the write that prefixes the CRC-covered body with its CRC.
3//!
4//! The size calculation sits beside the write because both walk the same
5//! optional fields, the v1-only timestamp and the two nullable byte strings,
6//! and a frame whose `message_size` disagrees with the bytes written is
7//! unparseable.
8
9use {
10    super::{Magic, Message},
11    bytes::{BufMut, Bytes},
12};
13
14impl Message {
15    /// Number of bytes that the message occupies inside the per-entry frame.
16    ///
17    /// The count starts at the CRC field. It goes up to and includes the
18    /// value.
19    #[must_use]
20    pub fn encoded_len(&self) -> usize {
21        // crc(4) + magic(1) + attrs(1) [+ ts(8) if v1] + key + value
22        let mut n = 4 + 1 + 1;
23        if matches!(self.magic, Magic::V1) {
24            n += 8;
25        }
26        n += nullable_bytes_len(self.key.as_ref());
27        n += nullable_bytes_len(self.value.as_ref());
28        n
29    }
30
31    /// Encode the message into `buf`, including the CRC field.
32    ///
33    /// This method extends `buf`. It writes nothing before the CRC.
34    pub fn encode_into<B: BufMut>(&self, buf: &mut B) {
35        // Build the CRC-covered payload first, then prefix it with the CRC.
36        // `encoded_len` counts the 4-byte CRC this body excludes; reserving it
37        // anyway costs four bytes and keeps arithmetic out of a capacity hint,
38        // where a wrong answer is invisible and so untestable.
39        let mut body = Vec::with_capacity(self.encoded_len());
40        body.push(self.magic.as_i8().cast_unsigned());
41        body.push(self.attributes.cast_unsigned());
42        if matches!(self.magic, Magic::V1) {
43            let ts = self.timestamp.unwrap_or(-1);
44            body.extend_from_slice(&ts.to_be_bytes());
45        }
46        put_nullable_bytes(&mut body, self.key.as_ref());
47        put_nullable_bytes(&mut body, self.value.as_ref());
48
49        let crc = crc32fast::hash(&body);
50        buf.put_u32(crc);
51        buf.put_slice(&body);
52    }
53}
54
55fn nullable_bytes_len(b: Option<&Bytes>) -> usize { 4 + b.map_or(0, Bytes::len) }
56
57fn put_nullable_bytes(buf: &mut Vec<u8>, b: Option<&Bytes>) {
58    match b {
59        None => buf.extend_from_slice(&(-1i32).to_be_bytes()),
60        Some(data) => {
61            let len = i32::try_from(data.len()).unwrap_or(i32::MAX);
62            buf.extend_from_slice(&len.to_be_bytes());
63            buf.extend_from_slice(data);
64        }
65    }
66}
67
68#[cfg(test)]
69mod tests {
70
71    use {
72        super::*,
73        crate::message::test_support::{
74            RoundTrip, fixture_v0, fixture_v1, fixture_v1_null, round_trip,
75        },
76    };
77
78    #[yare::parameterized(
79        v0 = { fixture_v0() },
80        v1 = { fixture_v1() },
81    )]
82    fn message_roundtrips(message: Message) {
83        let RoundTrip { buffer, decoded } = round_trip(&message);
84        assert2::assert!(buffer.len() == message.encoded_len());
85        assert2::assert!(decoded == message);
86    }
87
88    #[test]
89    fn v1_null_key_and_value() {
90        let m = fixture_v1_null();
91        let RoundTrip { decoded, .. } = round_trip(&m);
92        assert2::assert!(decoded == m);
93    }
94
95    #[test]
96    fn v1_missing_timestamp_encodes_minus_one() {
97        let m = Message {
98            magic: Magic::V1,
99            attributes: 0,
100            timestamp: None,
101            key: None,
102            value: None,
103        };
104        let RoundTrip { decoded, .. } = round_trip(&m);
105        assert2::assert!(
106            decoded
107                == Message {
108                    timestamp: Some(-1),
109                    ..m
110                }
111        );
112    }
113
114    #[test]
115    fn empty_key_is_some_not_null() {
116        // len == 0 is an empty (non-null) field; only len == -1 is null.
117        let m = Message {
118            magic: Magic::V0,
119            attributes: 0,
120            timestamp: None,
121            key: Some(Bytes::new()),
122            value: Some(Bytes::from_static(b"v")),
123        };
124        let RoundTrip { decoded, .. } = round_trip(&m);
125        assert2::assert!(decoded == m);
126    }
127}
krabka_records_legacy/message/encode.rs

krabka_records_legacy/message/
encode.rs
