osdp/
multipart.rs

1//! Multi-part message engine.
2//!
3//! # Spec: §5.10, Table 4
4//!
5//! When a command or reply payload is larger than a single packet supports,
6//! the data block is prefixed with a 6-byte multi-part header:
7//!
8//! ```text
9//! +---------------+--------------+--------------------+
10//! | MpSizeTotal   | MpOffset     | MpFragmentSize     |
11//! | (u16 LE)      | (u16 LE)     | (u16 LE)           |
12//! +---------------+--------------+--------------------+
13//! ```
14//!
15//! Rules enforced by this module:
16//! - First fragment is always at offset 0.
17//! - Fragments are strictly sequential (no gaps, no reorder).
18//! - All fragments report the same `MpSizeTotal`.
19//! - Sender may abort by setting `MpOffset >= MpSizeTotal` and
20//!   `MpFragmentSize = 0`.
21//!
22//! See [`flow`] for a rendered fragment-exchange diagram.
23
24use crate::error::{Error, MultipartError};
25
26/// Rendered diagram of a multi-part transfer.
27///
28/// Each fragment is a normal OSDP packet whose data block is prefixed with a
29/// [`MultipartHeader`]. The receiver tracks the next-expected offset and
30/// rejects any out-of-order, overlapping, or total-changing fragment.
31///
32#[cfg_attr(feature = "_docs", aquamarine::aquamarine)]
33/// ```mermaid
34/// sequenceDiagram
35///     participant TX as MultipartTx<br/>(splitter)
36///     participant RX as MultipartRx<br/>(assembler)
37///     TX->>RX: hdr(total=N, offset=0,         frag=k₀) ‖ payload₀
38///     RX->>RX: state = AwaitOffset(k₀)
39///     TX->>RX: hdr(total=N, offset=k₀,        frag=k₁) ‖ payload₁
40///     RX->>RX: state = AwaitOffset(k₀+k₁)
41///     TX->>RX: hdr(total=N, offset=k₀+k₁,     frag=k₂) ‖ payload₂
42///     RX-->>TX: complete (reassembled body, N bytes)
43///     Note over TX,RX: TX may abort by sending<br/>hdr(total=N, offset≥N, frag=0)
44/// ```
45#[cfg(feature = "alloc")]
46pub mod flow {}
47
48/// Multi-part header on the wire.
49#[derive(Debug, Clone, Copy, PartialEq, Eq)]
50pub struct MultipartHeader {
51    /// Total combined size of all fragments.
52    pub total: u16,
53    /// Byte offset of this fragment's data within the combined transfer.
54    pub offset: u16,
55    /// Length of this fragment's data.
56    pub fragment: u16,
57}
58
59impl MultipartHeader {
60    /// Encoded size on the wire.
61    pub const WIRE_LEN: usize = 6;
62
63    /// Encode into a 6-byte array.
64    pub const fn encode(self) -> [u8; Self::WIRE_LEN] {
65        let t = self.total.to_le_bytes();
66        let o = self.offset.to_le_bytes();
67        let f = self.fragment.to_le_bytes();
68        [t[0], t[1], o[0], o[1], f[0], f[1]]
69    }
70
71    /// Decode from a 6-byte array.
72    pub fn decode(bytes: &[u8]) -> Result<Self, Error> {
73        if bytes.len() < Self::WIRE_LEN {
74            return Err(Error::Truncated {
75                have: bytes.len(),
76                need: Self::WIRE_LEN,
77            });
78        }
79        Ok(Self {
80            total: u16::from_le_bytes([bytes[0], bytes[1]]),
81            offset: u16::from_le_bytes([bytes[2], bytes[3]]),
82            fragment: u16::from_le_bytes([bytes[4], bytes[5]]),
83        })
84    }
85
86    /// `true` if the header signals an abort.
87    pub const fn is_abort(self) -> bool {
88        self.fragment == 0 && self.offset >= self.total
89    }
90}
91
92#[cfg(feature = "alloc")]
93mod alloc_impls {
94    use super::*;
95    use alloc::vec::Vec;
96
97    /// Iterator-style splitter for outgoing multi-part transfers.
98    #[derive(Debug, Clone)]
99    pub struct MultipartTx<'a> {
100        body: &'a [u8],
101        offset: u16,
102        fragment_size: u16,
103    }
104
105    impl<'a> MultipartTx<'a> {
106        /// New splitter. `fragment_size` is the maximum payload size that
107        /// fits into a single packet (excluding the 6-byte multi-part
108        /// header).
109        ///
110        /// # Example
111        ///
112        /// ```
113        /// use osdp::multipart::{MultipartHeader, MultipartTx};
114        /// let body = b"hello, multi-part world";
115        /// let tx = MultipartTx::new(body, 8)?;
116        /// let parts: Vec<_> = tx.collect();
117        /// assert_eq!(parts.len(), 3);
118        /// assert_eq!(parts[0].0.total, body.len() as u16);
119        /// assert_eq!(parts[0].0.offset, 0);
120        /// assert_eq!(parts[0].1, &body[..8]);
121        /// # Ok::<(), osdp::Error>(())
122        /// ```
123        pub fn new(body: &'a [u8], fragment_size: u16) -> Result<Self, Error> {
124            if fragment_size == 0 {
125                return Err(Error::Multipart(MultipartError::BadFragmentSize));
126            }
127            if body.len() > u16::MAX as usize {
128                return Err(Error::BufferOverflow {
129                    need: body.len(),
130                    have: u16::MAX as usize,
131                });
132            }
133            Ok(Self {
134                body,
135                offset: 0,
136                fragment_size,
137            })
138        }
139
140        /// Total transfer size.
141        pub fn total(&self) -> u16 {
142            self.body.len() as u16
143        }
144
145        /// Bytes still queued.
146        pub fn remaining(&self) -> u16 {
147            self.total().saturating_sub(self.offset)
148        }
149    }
150
151    impl<'a> Iterator for MultipartTx<'a> {
152        type Item = (MultipartHeader, &'a [u8]);
153
154        fn next(&mut self) -> Option<Self::Item> {
155            if self.offset as usize >= self.body.len() {
156                return None;
157            }
158            let take =
159                (self.body.len() - self.offset as usize).min(self.fragment_size as usize) as u16;
160            let frag = &self.body[self.offset as usize..self.offset as usize + take as usize];
161            let header = MultipartHeader {
162                total: self.body.len() as u16,
163                offset: self.offset,
164                fragment: take,
165            };
166            self.offset += take;
167            Some((header, frag))
168        }
169    }
170
171    /// Reassembler for incoming multi-part transfers.
172    #[derive(Debug, Clone, Default)]
173    pub struct MultipartRx {
174        buf: Vec<u8>,
175        total: Option<u16>,
176    }
177
178    impl MultipartRx {
179        /// New empty reassembler.
180        pub fn new() -> Self {
181            Self::default()
182        }
183
184        /// Reset state, dropping any partial transfer.
185        pub fn reset(&mut self) {
186            self.buf.clear();
187            self.total = None;
188        }
189
190        /// Feed one fragment. Returns `Ok(Some(_))` once the transfer is
191        /// complete (and resets internal state), `Ok(None)` if more
192        /// fragments are required.
193        pub fn push(
194            &mut self,
195            header: MultipartHeader,
196            data: &[u8],
197        ) -> Result<Option<Vec<u8>>, Error> {
198            if header.is_abort() {
199                self.reset();
200                return Err(Error::Multipart(MultipartError::Aborted));
201            }
202
203            if header.fragment as usize != data.len() {
204                return Err(Error::Multipart(MultipartError::BadFragmentSize));
205            }
206
207            match self.total {
208                None => {
209                    if header.offset != 0 {
210                        return Err(Error::Multipart(MultipartError::UnexpectedFirstOffset(
211                            header.offset,
212                        )));
213                    }
214                    self.total = Some(header.total);
215                    self.buf.clear();
216                    self.buf.reserve(header.total as usize);
217                }
218                Some(t) if t != header.total => {
219                    return Err(Error::Multipart(MultipartError::InconsistentTotal {
220                        first: t,
221                        now: header.total,
222                    }));
223                }
224                _ => {
225                    if header.offset as usize != self.buf.len() {
226                        return Err(Error::Multipart(MultipartError::OutOfOrderOffset {
227                            expected: self.buf.len() as u16,
228                            got: header.offset,
229                        }));
230                    }
231                }
232            }
233
234            if (header.offset as usize) + data.len() > header.total as usize {
235                return Err(Error::Multipart(MultipartError::OverflowsTotal));
236            }
237
238            self.buf.extend_from_slice(data);
239
240            if self.buf.len() == header.total as usize {
241                let total = header.total;
242                let mut out = core::mem::take(&mut self.buf);
243                self.total = None;
244                out.truncate(total as usize);
245                Ok(Some(out))
246            } else {
247                Ok(None)
248            }
249        }
250
251        /// Bytes received so far (only meaningful while a transfer is in progress).
252        pub fn progress(&self) -> usize {
253            self.buf.len()
254        }
255    }
256
257    #[cfg(test)]
258    mod tests {
259        use super::*;
260
261        #[test]
262        fn split_then_assemble() {
263            let body: Vec<u8> = (0u16..1024).flat_map(|n| n.to_le_bytes()).collect();
264            let tx = MultipartTx::new(&body, 200).unwrap();
265            let mut rx = MultipartRx::new();
266            let mut out = None;
267            for (h, frag) in tx {
268                if let Some(v) = rx.push(h, frag).unwrap() {
269                    out = Some(v);
270                }
271            }
272            assert_eq!(out.unwrap(), body);
273        }
274
275        #[test]
276        fn reject_out_of_order() {
277            let mut rx = MultipartRx::new();
278            let h0 = MultipartHeader {
279                total: 8,
280                offset: 0,
281                fragment: 4,
282            };
283            rx.push(h0, &[0u8; 4]).unwrap();
284            let bogus = MultipartHeader {
285                total: 8,
286                offset: 6,
287                fragment: 2,
288            };
289            assert!(rx.push(bogus, &[0u8; 2]).is_err());
290        }
291
292        #[test]
293        fn reject_inconsistent_total() {
294            let mut rx = MultipartRx::new();
295            let h0 = MultipartHeader {
296                total: 8,
297                offset: 0,
298                fragment: 4,
299            };
300            rx.push(h0, &[0u8; 4]).unwrap();
301            let bogus = MultipartHeader {
302                total: 12,
303                offset: 4,
304                fragment: 4,
305            };
306            assert!(rx.push(bogus, &[0u8; 4]).is_err());
307        }
308
309        #[test]
310        fn abort_signal() {
311            let mut rx = MultipartRx::new();
312            let h0 = MultipartHeader {
313                total: 8,
314                offset: 0,
315                fragment: 4,
316            };
317            rx.push(h0, &[0u8; 4]).unwrap();
318            let abort = MultipartHeader {
319                total: 8,
320                offset: 8,
321                fragment: 0,
322            };
323            assert!(matches!(
324                rx.push(abort, &[]),
325                Err(Error::Multipart(MultipartError::Aborted))
326            ));
327        }
328
329        #[test]
330        fn header_roundtrip() {
331            let h = MultipartHeader {
332                total: 0x1234,
333                offset: 0x0050,
334                fragment: 0x0010,
335            };
336            let bytes = h.encode();
337            assert_eq!(MultipartHeader::decode(&bytes).unwrap(), h);
338        }
339    }
340}
341
342#[cfg(feature = "alloc")]
343pub use alloc_impls::{MultipartRx, MultipartTx};
osdp/multipart.rs

osdp/
multipart.rs
