bacnet_mstp_codec/stream.rs
1//! Incremental decoding over caller-owned storage.
2
3use core::fmt;
4
5use crate::decode::{check_frame, decode_extended_in_place};
6use crate::frame::{DATA_OFFSET, wire_length};
7use crate::{Frame, FrameError, FrameHeader, PREAMBLE};
8
9/// The outcome of one [`StreamDecoder::push_slice`] call.
10#[derive(Debug, Clone, Copy, PartialEq, Eq)]
11#[must_use = "the event may be a frame, which the next push drops"]
12pub enum DecodeEvent<'a> {
13 /// Nothing complete and nothing to discard: push more octets.
14 NeedMore,
15 /// A valid frame. It borrows the decoder's storage and stays buffered
16 /// until the next push, which removes it.
17 Frame {
18 /// The decoded frame.
19 frame: Frame<'a>,
20 /// The frame's wire length.
21 consumed: usize,
22 },
23 /// Line noise was discarded while searching for a preamble: every
24 /// buffered octet up to the next `0x55` among them, which might start
25 /// one, in a single event. A run of noise that arrives over several
26 /// pushes gives one event per push.
27 ///
28 /// This is not a received invalid frame in terms of the Clause 9.5.4
29 /// receive state machine.
30 Invalid {
31 /// Octets discarded.
32 discard: usize,
33 },
34 /// A candidate frame failed a check and was dropped.
35 Malformed {
36 /// Octets of the stream discarded, counted from the candidate's
37 /// first octet.
38 ///
39 /// When its header passed every check, its declared length can be
40 /// trusted, so the decoder drops the whole frame by count, as the
41 /// receive state machine skips one (Clause 9.5.4.4): for a frame
42 /// longer than the storage, a reserved COBS type, or data that failed
43 /// its CRC. This is then the frame's wire length, and any of it that
44 /// hasn't arrived yet is dropped as it comes, with no further event.
45 /// For anything else, a bad header or one that never completes, it is
46 /// the first preamble octet, and the octets after it are scanned
47 /// again.
48 discard: usize,
49 /// Why the candidate was dropped.
50 error: FrameError,
51 /// The candidate's destination address once its header CRC has
52 /// passed, including for a header refused afterwards (source 255, a
53 /// Length out of range). `None` when the header CRC failed or the
54 /// header never completed. See [`StreamDecoder`] for why a receiver
55 /// needs it.
56 destination: Option<u8>,
57 },
58 /// The pushed slice did not fit in the free storage. None of it was
59 /// taken; see [`StreamDecoder`].
60 Error(FrameError),
61}
62
63/// Receive phase represented by the decoder's retained bytes.
64#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
65pub enum StreamPhase {
66 /// Searching for the first preamble octet; retained bytes are only noise.
67 Sync,
68 /// The first preamble octet has been recognized.
69 Preamble,
70 /// Both preamble octets are present and the fixed header is incomplete.
71 Header,
72 /// A valid fixed header is present and the rest of its frame hasn't all
73 /// arrived, whether the decoder is buffering the frame or skipping it by
74 /// count.
75 Data,
76}
77
78/// Incremental frame decoder over a buffer the caller lends it.
79///
80/// Feed it received octets with [`push`](Self::push) or
81/// [`push_slice`](Self::push_slice). Each call buffers the input and reports
82/// at most one [`DecodeEvent`], so:
83///
84/// - A [`DecodeEvent::Frame`] borrows the storage. The frame stays buffered
85/// until the next push, which commits it (drops it from the buffer) before
86/// it reads new input. An empty push commits it too.
87/// - After any event other than [`DecodeEvent::NeedMore`], more may already be
88/// buffered, such as a second frame from the same slice or more noise. Call
89/// `push_slice(&[])` until it returns `NeedMore` to drain it.
90/// - A slice longer than the free storage is refused whole with
91/// [`DecodeEvent::Error`] carrying [`FrameError::BufferExhausted`] (the
92/// slice's length and the free room), and none of it is taken. Drain, then
93/// push it again or in smaller pieces. Octets that finish a frame being
94/// skipped by count need no room, and don't count toward the slice's
95/// length.
96///
97/// An extended frame is decoded in place, so the frame's data is its decoded
98/// NPDU, and the storage it was received in holds that from the octet after
99/// its header.
100///
101/// # Storage and skipping
102///
103/// Size the storage for the largest frame you accept, plus room for the
104/// slices you push: [`MAX_STANDARD_FRAME_LENGTH`](crate::MAX_STANDARD_FRAME_LENGTH)
105/// for every standard frame, or
106/// [`MAX_EXTENDED_FRAME_LENGTH`](crate::MAX_EXTENDED_FRAME_LENGTH) for
107/// extended frames too (see the [crate docs](crate)). A candidate that could
108/// never fit is reported once, as [`DecodeEvent::Malformed`] with
109/// [`FrameError::FrameTooLong`], as soon as its header shows its size, and
110/// is then skipped by count, so the decoder never wedges. Storage too small
111/// for even a header reports a candidate at its first octet and drops that
112/// octet.
113///
114/// Once a header passes its CRC and the header checks of Clause 9.5.8, its
115/// Length can be trusted: a frame dropped after that point (too long for the
116/// storage, a reserved COBS type, or data that fails its CRC) is skipped
117/// whole, as the receive state machine skips one (Clause 9.5.4.4), and
118/// nothing inside it can come out as a frame. A header that fails its CRC or
119/// those checks says nothing reliable about the frame's length, so only its
120/// first octet is dropped and the octets after it are scanned again. A
121/// decoder built [`with_reserved_cobs`](Self::with_reserved_cobs) buffers a
122/// reserved COBS type and checks it like an extended frame instead.
123///
124/// # Destinations
125///
126/// The decoder does not know which station it serves, so it reports every
127/// candidate it drops, whatever the destination. A receive state machine
128/// (Clause 9.5.4.4) only counts an oversized, unsupported or bad-data-CRC
129/// frame as invalid when it is addressed to that station or broadcast, and
130/// skips one meant for another station without complaint. Such a caller
131/// should check the `destination` of a [`DecodeEvent::Malformed`] carrying
132/// [`FrameError::FrameTooLong`], [`FrameError::UnsupportedCobsFrame`] or
133/// [`FrameError::DataCrcMismatch`] before treating it as an invalid frame. A
134/// bad header (a CRC failure, source 255, a Length out of range) counts
135/// wherever it was going. Since the header checks come before the type
136/// check, an unsupported COBS type always has a good header.
137///
138/// The decoder keeps no time. A receiver that aborts a partial frame after a
139/// silence (Clause 9.5.4) calls [`reset`](Self::reset), which also ends a
140/// skip.
141///
142/// ```
143/// use bacnet_mstp_codec::{DecodeEvent, Frame, FrameType, StreamDecoder};
144///
145/// let mut wire = [0u8; 16];
146/// Frame::new(FrameType::TOKEN, 1, 0, &[]).encode_into(&mut wire[..8])?;
147/// Frame::new(FrameType::TOKEN, 2, 1, &[]).encode_into(&mut wire[8..])?;
148///
149/// let mut storage = [0u8; 64];
150/// let mut decoder = StreamDecoder::new(&mut storage);
151/// let mut destinations = Vec::new();
152/// let mut event = decoder.push_slice(&wire);
153/// loop {
154/// match event {
155/// DecodeEvent::NeedMore => break,
156/// DecodeEvent::Frame { frame, .. } => destinations.push(frame.header.destination),
157/// _ => {}
158/// }
159/// // Commit the frame, if any, and look for the next one.
160/// event = decoder.push_slice(&[]);
161/// }
162/// assert_eq!(destinations, [1, 2]);
163/// assert_eq!(decoder.buffered_len(), 0);
164/// # Ok::<(), bacnet_mstp_codec::FrameError>(())
165/// ```
166pub struct StreamDecoder<'a> {
167 storage: &'a mut [u8],
168 /// Where the buffered octets begin. Octets the decoder is done with are
169 /// dropped by moving this past them, not by shifting the rest down, and
170 /// it goes back to 0 whenever nothing is buffered.
171 start: usize,
172 /// How many octets are buffered, from `start`.
173 length: usize,
174 /// Wire length of the frame the last push delivered, which the next
175 /// push commits.
176 pending: usize,
177 /// Octets of a dropped frame still to come, which pushes drop by count.
178 /// While it is nonzero nothing is buffered.
179 skip: usize,
180 /// Reserved COBS types are checked and yielded, not skipped; see
181 /// [`Self::with_reserved_cobs`].
182 reserved_cobs: bool,
183}
184
185/// The decoder's counts, phase and mode; the storage's contents are left out.
186impl fmt::Debug for StreamDecoder<'_> {
187 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
188 f.debug_struct("StreamDecoder")
189 .field("capacity", &self.capacity())
190 .field("buffered", &self.length)
191 .field("pending", &self.pending)
192 .field("skip", &self.skip)
193 .field("phase", &self.phase())
194 .field("reserved_cobs", &self.reserved_cobs)
195 .finish()
196 }
197}
198
199/// Offset of the destination octet: after the preamble and the type octet.
200const DESTINATION_INDEX: usize = PREAMBLE.len() + 1;
201/// Offset of the Length field's two octets, most significant first.
202const LENGTH_INDEX: usize = PREAMBLE.len() + 3;
203
204#[derive(Debug, Clone, Copy, PartialEq, Eq)]
205enum InternalEvent {
206 NeedMore,
207 Frame {
208 header: FrameHeader,
209 data_length: usize,
210 consumed: usize,
211 },
212 Invalid {
213 discard: usize,
214 },
215 Malformed {
216 discard: usize,
217 error: FrameError,
218 destination: Option<u8>,
219 },
220}
221
222impl<'a> StreamDecoder<'a> {
223 /// Create an empty decoder over `storage`.
224 #[must_use]
225 pub const fn new(storage: &'a mut [u8]) -> Self {
226 Self {
227 storage,
228 start: 0,
229 length: 0,
230 pending: 0,
231 skip: 0,
232 reserved_cobs: false,
233 }
234 }
235
236 /// Check the reserved COBS-encoded types, 34-127, as a receiver that
237 /// supports COBS-encoded frames does, rather than skip them.
238 ///
239 /// Such a receiver takes the encoded fields of every COBS type in, and
240 /// a frame whose CRC-32K checks out is a valid frame of a type it
241 /// doesn't know (Clause 9.5.4.9), for its state machine to ignore or
242 /// treat as unexpected. So a reserved type whose header passes is
243 /// buffered like an extended frame: when it fits the storage and its
244 /// encoded fields check out, it comes out as a [`DecodeEvent::Frame`]
245 /// with its type, which no [`FrameType`](crate::FrameType) constant
246 /// names, and its decoded data, and otherwise it is dropped and skipped
247 /// by count, with [`FrameError::DataCrcMismatch`] or
248 /// [`FrameError::FrameTooLong`].
249 /// [`FrameError::UnsupportedCobsFrame`] never comes out.
250 ///
251 /// A decoder from [`Self::new`] skips them by count as soon as their
252 /// header passes, as a receiver without COBS support does, and reports
253 /// [`FrameError::UnsupportedCobsFrame`].
254 #[must_use]
255 pub const fn with_reserved_cobs(mut self) -> Self {
256 self.reserved_cobs = true;
257 self
258 }
259
260 /// Whether the decoder checks reserved COBS types; see
261 /// [`Self::with_reserved_cobs`].
262 #[must_use]
263 pub const fn checks_reserved_cobs(&self) -> bool {
264 self.reserved_cobs
265 }
266
267 /// Number of buffered octets, including a frame awaiting the next push.
268 /// Octets of a frame being skipped by count are never buffered.
269 #[must_use]
270 pub const fn buffered_len(&self) -> usize {
271 self.length
272 }
273
274 /// Capacity of the caller-owned receive buffer.
275 #[must_use]
276 pub const fn capacity(&self) -> usize {
277 self.storage.len()
278 }
279
280 /// Return the semantic receive phase, excluding retained resync noise.
281 #[must_use]
282 pub fn phase(&self) -> StreamPhase {
283 if self.skip != 0 {
284 return StreamPhase::Data;
285 }
286 let buffered = self.buffered();
287 if self.pending != 0 || buffered.first() != Some(&PREAMBLE[0]) {
288 return StreamPhase::Sync;
289 }
290 if self.length == 1 {
291 return StreamPhase::Preamble;
292 }
293 if buffered[1] != PREAMBLE[1] {
294 return StreamPhase::Sync;
295 }
296 if self.length < DATA_OFFSET {
297 StreamPhase::Header
298 } else {
299 StreamPhase::Data
300 }
301 }
302
303 /// Clear all buffered input, the pending frame and any skip.
304 pub fn reset(&mut self) {
305 self.start = 0;
306 self.length = 0;
307 self.pending = 0;
308 self.skip = 0;
309 }
310
311 /// Feed one octet; see [`push_slice`](Self::push_slice).
312 pub fn push(&mut self, byte: u8) -> DecodeEvent<'_> {
313 self.push_slice(core::slice::from_ref(&byte))
314 }
315
316 /// Commit a frame delivered by the previous push, take `input` (dropping
317 /// what finishes a frame being skipped and buffering the rest), and
318 /// report the first resulting event. The [type docs](Self) describe the
319 /// protocol.
320 pub fn push_slice(&mut self, input: &[u8]) -> DecodeEvent<'_> {
321 self.commit_pending();
322 let skipped = self.skip.min(input.len());
323 let rest = &input[skipped..];
324 let free = self.storage.len() - self.length;
325 if rest.len() > free {
326 return DecodeEvent::Error(FrameError::BufferExhausted {
327 needed: rest.len(),
328 available: free,
329 });
330 }
331 self.skip -= skipped;
332 if self.start + self.length + rest.len() > self.storage.len() {
333 self.compact();
334 }
335 let end = self.start + self.length;
336 self.storage[end..end + rest.len()].copy_from_slice(rest);
337 self.length += rest.len();
338 match self.try_decode() {
339 InternalEvent::NeedMore => DecodeEvent::NeedMore,
340 InternalEvent::Frame {
341 header,
342 data_length,
343 consumed,
344 } => DecodeEvent::Frame {
345 frame: Frame {
346 header,
347 data: &self.buffered()[DATA_OFFSET..DATA_OFFSET + data_length],
348 },
349 consumed,
350 },
351 InternalEvent::Invalid { discard } => DecodeEvent::Invalid { discard },
352 InternalEvent::Malformed {
353 discard,
354 error,
355 destination,
356 } => DecodeEvent::Malformed {
357 discard,
358 error,
359 destination,
360 },
361 }
362 }
363
364 fn commit_pending(&mut self) {
365 if self.pending == 0 {
366 return;
367 }
368 self.discard(self.pending);
369 self.pending = 0;
370 }
371
372 fn try_decode(&mut self) -> InternalEvent {
373 let buffered = self.buffered();
374 let Some(&first) = buffered.first() else {
375 return InternalEvent::NeedMore;
376 };
377 let may_start_preamble =
378 first == PREAMBLE[0] && buffered.get(1).is_none_or(|&second| second == PREAMBLE[1]);
379 if !may_start_preamble {
380 // Noise, up to the next octet that might start a preamble.
381 let discard = buffered[1..]
382 .iter()
383 .position(|&octet| octet == PREAMBLE[0])
384 .map_or(buffered.len(), |index| index + 1);
385 self.discard(discard);
386 return InternalEvent::Invalid { discard };
387 }
388 let error = match check_frame(buffered, self.reserved_cobs) {
389 Ok(valid) => {
390 let end = self.start + self.length;
391 decode_extended_in_place(&mut self.storage[self.start..end], &valid);
392 self.pending = valid.consumed;
393 return InternalEvent::Frame {
394 header: valid.header,
395 data_length: valid.data_length,
396 consumed: valid.consumed,
397 };
398 }
399 Err(error) => error,
400 };
401 // Errors past the header CRC come with the whole header buffered,
402 // so its destination octet can be trusted. Every variant is listed,
403 // so a new error has to say which side of the CRC it falls on.
404 let destination = match error {
405 FrameError::TooShort { .. }
406 | FrameError::InvalidPreamble { .. }
407 | FrameError::HeaderCrcMismatch => None,
408 FrameError::BroadcastSource
409 | FrameError::InvalidCobsLength { .. }
410 | FrameError::Oversize { .. }
411 | FrameError::UnsupportedCobsFrame { .. }
412 | FrameError::Truncated { .. }
413 | FrameError::DataCrcMismatch { .. } => Some(self.buffered()[DESTINATION_INDEX]),
414 // check_frame never returns these: they come from decode_frame,
415 // from encoding or from the stream decoder itself.
416 FrameError::ExtendedFrame { .. }
417 | FrameError::UnexpectedData { .. }
418 | FrameError::MissingVendorId { .. }
419 | FrameError::FrameTooLong { .. }
420 | FrameError::BufferExhausted { .. } => None,
421 };
422 match error {
423 FrameError::TooShort { needed, .. } => self.wait_for_header(needed),
424 FrameError::Truncated { length, .. } => {
425 let needed = wire_length(length);
426 let capacity = self.storage.len();
427 if needed <= capacity {
428 return InternalEvent::NeedMore;
429 }
430 self.skip_frame(
431 needed,
432 FrameError::FrameTooLong { needed, capacity },
433 destination,
434 )
435 }
436 FrameError::UnsupportedCobsFrame { .. } => {
437 let length = self.declared_length();
438 self.skip_frame(wire_length(length), error, destination)
439 }
440 FrameError::DataCrcMismatch { length } => {
441 self.skip_frame(wire_length(length), error, destination)
442 }
443 _ => {
444 self.discard(1);
445 InternalEvent::Malformed {
446 discard: 1,
447 error,
448 destination,
449 }
450 }
451 }
452 }
453
454 /// Wait for the rest of a header, unless the storage can't hold the
455 /// `needed` octets of one: then drop the first octet so the decoder
456 /// resynchronizes instead of filling up for good.
457 fn wait_for_header(&mut self, needed: usize) -> InternalEvent {
458 let capacity = self.storage.len();
459 if needed <= capacity {
460 return InternalEvent::NeedMore;
461 }
462 self.discard(1);
463 InternalEvent::Malformed {
464 discard: 1,
465 error: FrameError::FrameTooLong { needed, capacity },
466 destination: None,
467 }
468 }
469
470 /// The Length field of the header at the front, which is complete.
471 fn declared_length(&self) -> usize {
472 let buffered = self.buffered();
473 usize::from(u16::from_be_bytes([
474 buffered[LENGTH_INDEX],
475 buffered[LENGTH_INDEX + 1],
476 ]))
477 }
478
479 /// Drop the candidate at the front, whose good header declares `total`
480 /// octets in all, by count: what is buffered of it now, and the rest as
481 /// it arrives.
482 fn skip_frame(
483 &mut self,
484 total: usize,
485 error: FrameError,
486 destination: Option<u8>,
487 ) -> InternalEvent {
488 if total <= self.length {
489 self.discard(total);
490 } else {
491 self.skip = total - self.length;
492 self.start = 0;
493 self.length = 0;
494 }
495 InternalEvent::Malformed {
496 discard: total,
497 error,
498 destination,
499 }
500 }
501
502 /// The buffered octets.
503 fn buffered(&self) -> &[u8] {
504 &self.storage[self.start..self.start + self.length]
505 }
506
507 /// Drop the first `count` buffered octets, without moving the rest.
508 fn discard(&mut self, count: usize) {
509 let count = count.min(self.length);
510 self.start += count;
511 self.length -= count;
512 if self.length == 0 {
513 self.start = 0;
514 }
515 }
516
517 /// Move the buffered octets to the front of the storage, to make room
518 /// after them. They go a piece at a time, each piece no longer than the
519 /// free octets in front of it, so no copy overlaps its source: a plain
520 /// copy does each, not a `memmove`.
521 fn compact(&mut self) {
522 let gap = self.start;
523 if gap == 0 {
524 return;
525 }
526 let end = self.start + self.length;
527 let mut from = gap;
528 while from < end {
529 let count = gap.min(end - from);
530 let (front, back) = self.storage.split_at_mut(from);
531 front[from - gap..from - gap + count].copy_from_slice(&back[..count]);
532 from += count;
533 }
534 self.start = 0;
535 }
536}