dcp/binary/
stream.rs

1//! Binary stream chunk for streaming support.
2
3use crate::DCPError;
4
5/// Chunk flag constants
6#[allow(non_snake_case)]
7pub mod ChunkFlags {
8    pub const FIRST: u8 = 0b0001;
9    pub const CONTINUE: u8 = 0b0010;
10    pub const LAST: u8 = 0b0100;
11    pub const ERROR: u8 = 0b1000;
12}
13
14const ALL_CHUNK_FLAGS: u8 =
15    ChunkFlags::FIRST | ChunkFlags::CONTINUE | ChunkFlags::LAST | ChunkFlags::ERROR;
16
17fn valid_chunk_flags(flags: u8) -> bool {
18    if flags == 0 || flags & !ALL_CHUNK_FLAGS != 0 {
19        return false;
20    }
21    if flags & ChunkFlags::ERROR != 0 {
22        return flags == ChunkFlags::ERROR;
23    }
24    matches!(
25        flags,
26        ChunkFlags::FIRST | ChunkFlags::CONTINUE | ChunkFlags::LAST
27    ) || flags == (ChunkFlags::FIRST | ChunkFlags::LAST)
28}
29
30/// Binary stream chunk - 7 bytes header
31#[repr(C, packed)]
32#[derive(Debug, Clone, Copy, PartialEq, Eq)]
33pub struct StreamChunk {
34    /// Sequence number for ordering
35    pub sequence: u32,
36    /// Flags: FIRST=1, CONTINUE=2, LAST=4, ERROR=8
37    pub flags: u8,
38    /// Chunk payload length
39    pub len: u16,
40    // Payload follows
41}
42
43impl StreamChunk {
44    /// Size of the chunk header in bytes
45    pub const SIZE: usize = 7;
46
47    /// Create a new stream chunk
48    pub fn new(sequence: u32, flags: u8, len: u16) -> Self {
49        Self {
50            sequence,
51            flags,
52            len,
53        }
54    }
55
56    /// Create the first chunk in a stream
57    pub fn first(sequence: u32, len: u16) -> Self {
58        Self::new(sequence, ChunkFlags::FIRST, len)
59    }
60
61    /// Create a continuation chunk
62    pub fn continuation(sequence: u32, len: u16) -> Self {
63        Self::new(sequence, ChunkFlags::CONTINUE, len)
64    }
65
66    /// Create the last chunk in a stream
67    pub fn last(sequence: u32, len: u16) -> Self {
68        Self::new(sequence, ChunkFlags::LAST, len)
69    }
70
71    /// Create an error chunk
72    pub fn error(sequence: u32, len: u16) -> Self {
73        Self::new(sequence, ChunkFlags::ERROR, len)
74    }
75
76    /// Parse from bytes
77    #[inline(always)]
78    pub fn from_bytes(bytes: &[u8]) -> Result<&Self, DCPError> {
79        if bytes.len() < Self::SIZE {
80            return Err(DCPError::InsufficientData);
81        }
82        if !valid_chunk_flags(bytes[4]) {
83            return Err(DCPError::ValidationFailed);
84        }
85        // SAFETY: We've verified the slice is at least SIZE bytes
86        Ok(unsafe { &*(bytes.as_ptr() as *const Self) })
87    }
88
89    /// Serialize to bytes
90    #[inline(always)]
91    pub fn as_bytes(&self) -> &[u8] {
92        // SAFETY: StreamChunk is repr(C, packed) with no padding
93        unsafe { std::slice::from_raw_parts(self as *const Self as *const u8, Self::SIZE) }
94    }
95
96    /// Check if this is the first chunk
97    #[inline(always)]
98    pub fn is_first(&self) -> bool {
99        self.flags & ChunkFlags::FIRST != 0
100    }
101
102    /// Check if this is a continuation chunk
103    #[inline(always)]
104    pub fn is_continuation(&self) -> bool {
105        self.flags & ChunkFlags::CONTINUE != 0
106    }
107
108    /// Check if this is the last chunk
109    #[inline(always)]
110    pub fn is_last(&self) -> bool {
111        self.flags & ChunkFlags::LAST != 0
112    }
113
114    /// Check if this is an error chunk
115    #[inline(always)]
116    pub fn is_error(&self) -> bool {
117        self.flags & ChunkFlags::ERROR != 0
118    }
119}
120
121#[cfg(test)]
122mod tests {
123    use super::*;
124
125    #[test]
126    fn test_chunk_size() {
127        assert_eq!(std::mem::size_of::<StreamChunk>(), 7);
128    }
129
130    #[test]
131    fn test_chunk_round_trip() {
132        let chunk = StreamChunk::new(12345, ChunkFlags::FIRST | ChunkFlags::LAST, 1024);
133        let bytes = chunk.as_bytes();
134        let parsed = StreamChunk::from_bytes(bytes).unwrap();
135
136        // Copy values from packed struct before comparing
137        let sequence = parsed.sequence;
138        let flags = parsed.flags;
139        let len = parsed.len;
140
141        assert_eq!(sequence, 12345);
142        assert_eq!(flags, ChunkFlags::FIRST | ChunkFlags::LAST);
143        assert_eq!(len, 1024);
144    }
145
146    #[test]
147    fn test_flag_helpers() {
148        let first = StreamChunk::first(0, 100);
149        assert!(first.is_first());
150        assert!(!first.is_last());
151
152        let last = StreamChunk::last(1, 50);
153        assert!(last.is_last());
154        assert!(!last.is_first());
155
156        let error = StreamChunk::error(2, 0);
157        assert!(error.is_error());
158    }
159
160    #[test]
161    fn test_insufficient_data() {
162        let bytes = [0u8; 4];
163        assert_eq!(
164            StreamChunk::from_bytes(&bytes),
165            Err(DCPError::InsufficientData)
166        );
167    }
168}
dcp/binary/stream.rs

dcp/binary/
stream.rs
