exarrow_rs/transport/native/
framing.rs

1use std::sync::atomic::{AtomicU32, Ordering};
2
3use super::constants::HEADER_SIZE;
4use crate::error::TransportError;
5
6/// The 21-byte binary message header for the native protocol.
7///
8/// All multi-byte fields are in native (little-endian) byte order. The C++ SDK's
9/// `exaBswap32` helper is a no-op on little-endian platforms (`#ifdef WORDS_BIGENDIAN`),
10/// so the wire format is LE on all x86/x64 Exasol deployments.
11#[derive(Debug, Clone, PartialEq, Eq)]
12pub struct MessageHeader {
13    /// Payload size after the header.
14    pub message_length: u32,
15    /// Command code (e.g. CMD_EXECUTE, CMD_FETCH2).
16    pub command: u8,
17    /// Monotonically increasing message counter.
18    pub serial: u32,
19    /// Number of attributes following the header.
20    pub num_attributes: u32,
21    /// Total byte length of attribute data.
22    pub attribute_data_len: u32,
23    /// Number of data parts in the payload.
24    pub num_result_parts: u32,
25}
26
27impl MessageHeader {
28    pub fn new(
29        command: u8,
30        serial: u32,
31        num_attributes: u32,
32        attribute_data_len: u32,
33        num_result_parts: u32,
34    ) -> Self {
35        let message_length = attribute_data_len;
36        Self {
37            message_length,
38            command,
39            serial,
40            num_attributes,
41            attribute_data_len,
42            num_result_parts,
43        }
44    }
45
46    /// Serialize the header into a 21-byte array.
47    pub fn serialize(&self) -> [u8; HEADER_SIZE] {
48        let mut buf = [0u8; HEADER_SIZE];
49        buf[0..4].copy_from_slice(&self.message_length.to_le_bytes());
50        buf[4] = self.command;
51        buf[5..9].copy_from_slice(&self.serial.to_le_bytes());
52        buf[9..13].copy_from_slice(&self.num_attributes.to_le_bytes());
53        buf[13..17].copy_from_slice(&self.attribute_data_len.to_le_bytes());
54        buf[17..21].copy_from_slice(&self.num_result_parts.to_le_bytes());
55        buf
56    }
57
58    /// Parse a 21-byte array into a `MessageHeader`.
59    pub fn parse(data: &[u8; HEADER_SIZE]) -> Result<Self, TransportError> {
60        let message_length = u32::from_le_bytes([data[0], data[1], data[2], data[3]]);
61        let command = data[4];
62        let serial = u32::from_le_bytes([data[5], data[6], data[7], data[8]]);
63        let num_attributes = u32::from_le_bytes([data[9], data[10], data[11], data[12]]);
64        let attribute_data_len = u32::from_le_bytes([data[13], data[14], data[15], data[16]]);
65        let num_result_parts = u32::from_le_bytes([data[17], data[18], data[19], data[20]]);
66
67        Ok(Self {
68            message_length,
69            command,
70            serial,
71            num_attributes,
72            attribute_data_len,
73            num_result_parts,
74        })
75    }
76}
77
78/// Thread-safe monotonically increasing counter for message serial numbers.
79pub struct SerialCounter {
80    next: AtomicU32,
81}
82
83impl SerialCounter {
84    pub fn new() -> Self {
85        Self {
86            next: AtomicU32::new(1),
87        }
88    }
89
90    /// Return the next serial number, incrementing the internal counter.
91    pub fn next(&self) -> u32 {
92        self.next.fetch_add(1, Ordering::Relaxed)
93    }
94}
95
96impl Default for SerialCounter {
97    fn default() -> Self {
98        Self::new()
99    }
100}
101
102#[cfg(test)]
103mod tests {
104    use super::*;
105
106    #[test]
107    fn test_header_roundtrip() {
108        let header = MessageHeader {
109            message_length: 1024,
110            command: 12,
111            serial: 7,
112            num_attributes: 3,
113            attribute_data_len: 256,
114            num_result_parts: 1,
115        };
116        let bytes = header.serialize();
117        assert_eq!(bytes.len(), HEADER_SIZE);
118
119        let parsed = MessageHeader::parse(&bytes).unwrap();
120        assert_eq!(parsed, header);
121    }
122
123    #[test]
124    fn test_header_zero_values() {
125        let header = MessageHeader::new(0, 0, 0, 0, 0);
126        let bytes = header.serialize();
127        let parsed = MessageHeader::parse(&bytes).unwrap();
128        assert_eq!(parsed.message_length, 0);
129        assert_eq!(parsed.command, 0);
130        assert_eq!(parsed.serial, 0);
131        assert_eq!(parsed.num_attributes, 0);
132        assert_eq!(parsed.attribute_data_len, 0);
133        assert_eq!(parsed.num_result_parts, 0);
134    }
135
136    #[test]
137    fn test_header_max_values() {
138        let header = MessageHeader {
139            message_length: u32::MAX,
140            command: u8::MAX,
141            serial: u32::MAX,
142            num_attributes: u32::MAX,
143            attribute_data_len: u32::MAX,
144            num_result_parts: u32::MAX,
145        };
146        let bytes = header.serialize();
147        let parsed = MessageHeader::parse(&bytes).unwrap();
148        assert_eq!(parsed, header);
149    }
150
151    #[test]
152    fn test_serial_counter_increments() {
153        let counter = SerialCounter::new();
154        assert_eq!(counter.next(), 1);
155        assert_eq!(counter.next(), 2);
156        assert_eq!(counter.next(), 3);
157    }
158
159    #[test]
160    fn test_serial_counter_is_monotonic() {
161        let counter = SerialCounter::new();
162        let mut prev = counter.next();
163        for _ in 0..100 {
164            let curr = counter.next();
165            assert!(curr > prev);
166            prev = curr;
167        }
168    }
169}
exarrow_rs/transport/native/framing.rs

exarrow_rs/transport/native/
framing.rs
