nexosim_byte_utils/
decode.rs

1//! # Byte stream decoding utilities.
2//!
3//! This module contains primitives for byte stream decoding implementation.
4//!
5//! ## `ByteDecoderModel` model
6//!
7//! The [`ByteDecoderModel`] model accepts a stream of bytes at its inputs and
8//! outputs the parsed data. The type of the parsed data is expected to have a
9//! variant for decoding errors that are not ignored. This model is generic over
10//! [`BufDecoder`], which implements the decoding functionality. Method
11//! [`BufDecoder::decode`] operates on any implementer of the
12//! [`bytes::Buf`](https://docs.rs/bytes/latest/bytes/buf/trait.Buf.html) trait.
13//! The decoded result can have one of the following values:
14//! * [`BufDecoderResult::Empty`]: the buffer has been exhausted and ignored,
15//! * [`BufDecoderResult::Partial`]: the buffer has been exhausted and part of
16//!   messages parsed,
17//! * [`BufDecoderResult::Ignored`]: part of the buffer has been consumed and
18//!   ignored,
19//! * [`BufDecoderResult::Decoded`]: part of the buffer has been consumed and
20//!   decoded.
21//!
22//! The following example shows a model that produces a pulse for every `0xAA`
23//! byte in the input stream and ignores all other bytes.
24//!
25//! ```rust
26//! use bytes::Buf;
27//!
28//! use serde::{self, Deserialize, Serialize};
29//!
30//! use nexosim_byte_utils::decode::{BufDecoder, BufDecoderResult, ByteDecoderModel, ProtoByteDecoder};
31//!
32//! #[derive(Default, Serialize, Deserialize)]
33//! pub struct AaDecoder {}
34//!
35//! impl BufDecoder<()> for AaDecoder {
36//!     type DecoderEnv = ();
37//!
38//!     fn decode<B: Buf>(&mut self, buf: &mut B, _: &()) -> BufDecoderResult<()> {
39//!         while buf.has_remaining() {
40//!             if buf.get_u8() == 0xAA {
41//!                 return BufDecoderResult::Decoded(());
42//!             }
43//!         }
44//!         BufDecoderResult::Empty
45//!     }
46//! }
47//!
48//! // Decoder model prototype.
49//! pub type ProtoDecoder = ProtoByteDecoder<(), AaDecoder>;
50//!
51//! // Decoder model.
52//! pub type Decoder = ByteDecoderModel<(), AaDecoder>;
53//!
54//! let decoder = ProtoDecoder::default();
55//! ```
56//!
57//! ## `ByteDelimitedDecoder`
58//!
59//! [`ByteDelimitedDecoder`] can serve as a more complex example. In its
60//! simplest usage, it can decode data separated based on delimiter bytes. The
61//! following example shows how to generate a pulse for every byte sequence of
62//! the form `[0xFF, ..., 0xAA]`, where `...` is any non-empty sequence of
63//! bytes.
64//!
65//! ```rust
66//! use nexosim_byte_utils::decode::{ByteDelimitedDecoder, ByteDelimitedDecoderEnv, ProtoByteDecoder};
67//!
68//! let decoder = ProtoByteDecoder::new(
69//!     ByteDelimitedDecoder::<(), (), ()>::new(0xFF, 0xAA),
70//!     ByteDelimitedDecoderEnv::<(), (), ()>::new(|_, _| {}),
71//! );
72//! ```
73//!
74//! For a more realistic example, see the implementation of the KISS protocol
75//! decoder in the [`kiss_decoder`] module.
76use std::fmt;
77use std::marker::PhantomData;
78use std::ops::{Deref, DerefMut};
79
80use buf_list::BufList;
81
82use bytes::{Buf, Bytes};
83
84use serde::de::{self, DeserializeOwned, MapAccess, SeqAccess, Visitor};
85use serde::ser::SerializeStruct;
86use serde::{self, Deserialize, Deserializer, Serialize, Serializer};
87
88use nexosim::model::{Context, Model, ProtoModel};
89use nexosim::ports::Output;
90
91/// Buffer list wrapper implementing serialization.
92struct Buffer(BufList);
93
94impl Deref for Buffer {
95    type Target = BufList;
96
97    fn deref(&self) -> &BufList {
98        &self.0
99    }
100}
101
102impl DerefMut for Buffer {
103    fn deref_mut(&mut self) -> &mut BufList {
104        &mut self.0
105    }
106}
107
108impl Serialize for Buffer {
109    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
110    where
111        S: Serializer,
112    {
113        let mut buf: Vec<u8> = Vec::new();
114        for chunk in self.iter() {
115            buf.extend(chunk);
116        }
117        let mut state = serializer.serialize_struct("Buffer", 1)?;
118        state.serialize_field("buf", &buf)?;
119        state.end()
120    }
121}
122
123impl<'de> Deserialize<'de> for Buffer {
124    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
125    where
126        D: Deserializer<'de>,
127    {
128        #[derive(Deserialize)]
129        #[serde(field_identifier, rename_all = "lowercase")]
130        enum Field {
131            Buf,
132        }
133
134        struct BufferVisitor;
135
136        impl<'de> Visitor<'de> for BufferVisitor {
137            type Value = Buffer;
138
139            fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
140                formatter.write_str("struct Buffer")
141            }
142
143            fn visit_seq<V>(self, mut seq: V) -> Result<Buffer, V::Error>
144            where
145                V: SeqAccess<'de>,
146            {
147                let buf: Vec<u8> = seq
148                    .next_element()?
149                    .ok_or_else(|| de::Error::invalid_length(0, &self))?;
150                let mut list = BufList::new();
151                list.push_chunk(Bytes::from(buf));
152                Ok(Buffer(list))
153            }
154
155            fn visit_map<V>(self, mut map: V) -> Result<Buffer, V::Error>
156            where
157                V: MapAccess<'de>,
158            {
159                let mut buf = None;
160                while let Some(key) = map.next_key()? {
161                    match key {
162                        Field::Buf => {
163                            if buf.is_some() {
164                                return Err(de::Error::duplicate_field("buf"));
165                            }
166                            buf = Some(map.next_value()?);
167                        }
168                    }
169                }
170
171                let buf: Vec<u8> = buf.ok_or_else(|| de::Error::missing_field("buf"))?;
172                let mut list = BufList::new();
173                list.push_chunk(Bytes::from(buf));
174                Ok(Buffer(list))
175            }
176        }
177
178        const FIELDS: &[&str] = &["buf"];
179        deserializer.deserialize_struct("Buffer", FIELDS, BufferVisitor)
180    }
181}
182
183/// Buffer decoding result.
184#[derive(Copy, Clone, Debug, Eq, PartialEq)]
185pub enum BufDecoderResult<T> {
186    /// Input buffer consumed, nothing decoded.
187    Empty,
188    /// Input buffer consumed, message decoding in progress.
189    Partial,
190    /// Part of the input was ignored, there is more data.
191    Ignored,
192    /// Part of the input buffer is decoded, there may be more data.
193    Decoded(T),
194}
195
196/// Buffer decoder trait.
197pub trait BufDecoder<T>: Serialize + DeserializeOwned {
198    /// Decoder environment.
199    type DecoderEnv: Send;
200
201    /// Decodes part of the input buffer consuming it.
202    fn decode<B: Buf>(
203        &mut self,
204        buf: &mut B,
205        decoder_env: &Self::DecoderEnv,
206    ) -> BufDecoderResult<T>;
207}
208
209/// Byte stream decoder model prototype.
210pub struct ProtoByteDecoder<T: Clone + Send + 'static, D: BufDecoder<T> + Send + 'static> {
211    /// Decoded data.
212    pub data_out: Output<T>,
213
214    /// Decoder.
215    decoder: D,
216
217    /// Decoder environment.
218    decoder_env: D::DecoderEnv,
219}
220
221impl<T, D> ProtoByteDecoder<T, D>
222where
223    T: Clone + Send + 'static,
224    D: BufDecoder<T> + Send + 'static,
225{
226    /// Creates new byte stream decoder model.
227    pub fn new(decoder: D, decoder_env: D::DecoderEnv) -> Self {
228        Self {
229            data_out: Output::new(),
230            decoder,
231            decoder_env,
232        }
233    }
234
235    fn build(self) -> (ByteDecoderModel<T, D>, D::DecoderEnv) {
236        (
237            ByteDecoderModel {
238                data_out: self.data_out,
239                buf: Buffer(BufList::new()),
240                decoder: self.decoder,
241            },
242            self.decoder_env,
243        )
244    }
245}
246
247impl<T, D> Default for ProtoByteDecoder<T, D>
248where
249    T: Clone + Send + 'static,
250    D: BufDecoder<T> + Default + Send + 'static,
251    D::DecoderEnv: Default,
252{
253    fn default() -> Self {
254        Self::new(D::default(), D::DecoderEnv::default())
255    }
256}
257
258impl<T, D> ProtoModel for ProtoByteDecoder<T, D>
259where
260    T: Clone + Send + 'static,
261    D: BufDecoder<T> + Send + 'static,
262{
263    type Model = ByteDecoderModel<T, D>;
264
265    fn build(
266        self,
267        _: &mut nexosim::model::BuildContext<Self>,
268    ) -> (ByteDecoderModel<T, D>, D::DecoderEnv) {
269        self.build()
270    }
271}
272
273impl<T, D> fmt::Debug for ProtoByteDecoder<T, D>
274where
275    T: Clone + Send + 'static,
276    D: BufDecoder<T> + Send + 'static,
277{
278    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
279        f.debug_struct("ProtoByteDecoder").finish_non_exhaustive()
280    }
281}
282
283/// Byte stream decoder model.
284#[derive(Serialize, Deserialize)]
285#[serde(bound = "")]
286pub struct ByteDecoderModel<T: Clone + Send + 'static, D: BufDecoder<T> + Send + 'static> {
287    /// Decoded data.
288    pub data_out: Output<T>,
289
290    /// Internal buffer.
291    buf: Buffer,
292
293    /// Decoder.
294    decoder: D,
295}
296
297#[Model(type Env = D::DecoderEnv)]
298impl<T, D> ByteDecoderModel<T, D>
299where
300    T: Clone + Send + 'static,
301    D: BufDecoder<T> + Send + 'static,
302{
303    /// Input bytes -- input port.
304    pub async fn bytes_in(&mut self, data: Bytes, _: &Context<Self>, env: &mut D::DecoderEnv) {
305        self.buf.push_chunk(data);
306        loop {
307            match self.decoder.decode(&mut *self.buf, env) {
308                BufDecoderResult::Decoded(data) => self.data_out.send(data).await,
309                BufDecoderResult::Ignored => {}
310                _ => break,
311            }
312        }
313    }
314}
315
316impl<T, D> fmt::Debug for ByteDecoderModel<T, D>
317where
318    T: Clone + Send + 'static,
319    D: BufDecoder<T> + Send + 'static,
320{
321    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
322        f.debug_struct("ByteDecoderModel").finish_non_exhaustive()
323    }
324}
325
326/// Result of successful byte stream transformation.
327#[derive(Clone, Debug)]
328pub enum Transformed {
329    /// No bytes.
330    None,
331
332    /// One byte.
333    One(u8),
334
335    /// Many bytes.
336    Many(Bytes),
337}
338
339/// Result of byte stream transformation.
340type TransformResult<T> = Result<Transformed, T>;
341
342/// Transformer callback type.
343type TransformCallback<T, S> =
344    Box<dyn Fn(&mut S, &[u8], u8) -> TransformResult<T> + Send + 'static>;
345
346/// Decoder callback type.
347type DecodeCallback<T, S> = Box<dyn Fn(&mut S, &[u8]) -> T + Send + 'static>;
348
349/// Packet decoder environment.
350pub struct ByteDelimitedDecoderEnv<T, S, R>
351where
352    T: Clone + Send + 'static,
353    S: Send + Serialize + for<'a> Deserialize<'a>,
354    R: Send + Serialize + for<'a> Deserialize<'a>,
355{
356    transform: TransformCallback<T, S>,
357    decode: DecodeCallback<T, R>,
358}
359
360impl<T, S, R> ByteDelimitedDecoderEnv<T, S, R>
361where
362    T: Clone + Send + 'static,
363    S: Send + Serialize + for<'a> Deserialize<'a>,
364    R: Send + Serialize + for<'a> Deserialize<'a>,
365{
366    /// Creates new byte delimited decoder environment.
367    pub fn new<F>(decode: F) -> Self
368    where
369        F: Fn(&mut R, &[u8]) -> T + Send + 'static,
370    {
371        Self::with_transform(|_, _, c| TransformResult::Ok(Transformed::One(c)), decode)
372    }
373
374    /// Creates new byte delimited decoder environment with character transform
375    /// function.
376    pub fn with_transform<F, G>(transform: F, decode: G) -> Self
377    where
378        F: Fn(&mut S, &[u8], u8) -> TransformResult<T> + Send + 'static,
379        G: Fn(&mut R, &[u8]) -> T + Send + 'static,
380    {
381        Self {
382            transform: Box::new(transform),
383            decode: Box::new(decode),
384        }
385    }
386}
387
388impl<T, S, R> fmt::Debug for ByteDelimitedDecoderEnv<T, S, R>
389where
390    T: Clone + Send + 'static,
391    S: Send + Serialize + for<'a> Deserialize<'a>,
392    R: Send + Serialize + for<'a> Deserialize<'a>,
393{
394    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
395        f.debug_struct("ByteDelimitedDecoderEnv")
396            .finish_non_exhaustive()
397    }
398}
399
400/// Packet decoder.
401#[derive(Serialize, Deserialize)]
402pub struct ByteDelimitedDecoder<T, S, R>
403where
404    T: Clone + Send + 'static,
405    S: Send + Serialize,
406    R: Send + Serialize,
407{
408    /// Packet start delimiter.
409    start: u8,
410
411    /// Packet end delimiter.
412    end: u8,
413
414    /// Packet decoding is in progress.
415    is_decoding: bool,
416
417    /// Decoder buffer.
418    buf: Vec<u8>,
419
420    /// Stream transformer state.
421    transformer_state: S,
422
423    /// Stream decoder state.
424    decoder_state: R,
425
426    /// Phantom data.
427    _t: PhantomData<T>,
428}
429
430impl<T, S, R> ByteDelimitedDecoder<T, S, R>
431where
432    T: Clone + Send + 'static,
433    S: Default + Send + Serialize,
434    R: Default + Send + Serialize,
435{
436    /// Creates new packet decoder.
437    pub fn new(start: u8, end: u8) -> Self {
438        Self {
439            start,
440            end,
441            is_decoding: false,
442            buf: Vec::with_capacity(1024),
443            transformer_state: S::default(),
444            decoder_state: R::default(),
445            _t: PhantomData {},
446        }
447    }
448}
449
450impl<T, S, R> BufDecoder<T> for ByteDelimitedDecoder<T, S, R>
451where
452    T: Clone + Send + 'static,
453    S: Send + Serialize + for<'de> Deserialize<'de>,
454    R: Send + Serialize + for<'de> Deserialize<'de>,
455{
456    type DecoderEnv = ByteDelimitedDecoderEnv<T, S, R>;
457
458    fn decode<B: Buf>(
459        &mut self,
460        buf: &mut B,
461        decoder_env: &Self::DecoderEnv,
462    ) -> BufDecoderResult<T> {
463        loop {
464            if !self.is_decoding {
465                self.buf.clear();
466                while buf.has_remaining() && buf.chunk()[0] != self.start {
467                    buf.advance(1);
468                }
469                if !buf.has_remaining() {
470                    return BufDecoderResult::Empty;
471                }
472                buf.advance(1);
473                self.is_decoding = true;
474            }
475            while buf.has_remaining() && buf.chunk()[0] != self.end {
476                match (decoder_env.transform)(&mut self.transformer_state, &self.buf, buf.get_u8())
477                {
478                    Ok(result) => match result {
479                        Transformed::None => {}
480                        Transformed::One(byte) => self.buf.push(byte),
481                        Transformed::Many(bytes) => self.buf.extend(bytes),
482                    },
483                    Err(data) => {
484                        self.is_decoding = false;
485                        return BufDecoderResult::Decoded(data);
486                    }
487                }
488            }
489            if !buf.has_remaining() {
490                return BufDecoderResult::Partial;
491            }
492            self.is_decoding = false;
493            if !self.buf.is_empty() {
494                break;
495            }
496        }
497        buf.advance(1);
498        BufDecoderResult::Decoded((decoder_env.decode)(&mut self.decoder_state, &self.buf))
499    }
500}
501
502impl<
503    T: Clone + Send + 'static,
504    S: Send + Serialize + for<'de> Deserialize<'de>,
505    R: Send + Serialize + for<'de> Deserialize<'de>,
506> fmt::Debug for ByteDelimitedDecoder<T, S, R>
507{
508    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
509        f.debug_struct("ByteDelimitedDecoder")
510            .finish_non_exhaustive()
511    }
512}
513
514/// # KISS protocol decoder.
515///
516/// This module implements [KISS
517/// protocol](https://en.wikipedia.org/wiki/KISS_(amateur_radio_protocol)) data
518/// decoding KISS.
519///
520/// The following example shows a decoder that decodes every non-empty correctly
521/// encoded byte sequence as a pulse.
522///
523/// ```rust
524/// use nexosim_byte_utils::decode::kiss_decoder::{FromKiss, ProtoKissModel};
525///
526/// #[derive(Clone, Copy, Debug, Eq, PartialEq)]
527/// pub enum Data {
528///     Pulse,
529///     Aborted,
530/// }
531///
532/// impl FromKiss for Data {
533///     fn abort_variant(_: &[u8], _: u8) -> Self {
534///         Data::Aborted
535///     }
536/// }
537///
538/// pub fn decode(_: &mut (),_: &[u8]) -> Data {
539///     Data::Pulse
540/// }
541///
542/// let mut decoder = ProtoKissModel::<Data, ()>::new(decode);
543/// ```
544pub mod kiss_decoder {
545    use std::fmt;
546
547    use serde::{self, Deserialize, Serialize};
548
549    use nexosim::model::{self, BuildContext, ProtoModel};
550    use nexosim::ports::Output;
551
552    /// Byte delimiter.
553    pub const FEND: u8 = 0xC0;
554
555    /// Escape byte.
556    pub const FESC: u8 = 0xDB;
557
558    /// Transformed byte delimiter.
559    pub const TFEND: u8 = 0xDC;
560
561    /// Transformed escape byte.
562    pub const TFESC: u8 = 0xDD;
563
564    /// Trait for data that can be parsed from KISS protocol.
565    pub trait FromKiss {
566        /// Data variant parsed in case of message abort (i.e. wrong escape
567        /// sequence).
568        fn abort_variant(previous: &[u8], byte: u8) -> Self;
569    }
570
571    /// KISS protocol decoder model prototype.
572    pub struct ProtoKissModel<
573        T: FromKiss + Clone + Send + 'static,
574        S: Default + Send + Serialize + for<'de> Deserialize<'de> + 'static,
575        const GFEND: u8 = FEND,
576        const GFESC: u8 = FESC,
577        const GTFEND: u8 = TFEND,
578        const GTFESC: u8 = TFESC,
579    > {
580        /// Decoded data.
581        pub data_out: Output<T>,
582
583        /// Decoder model.
584        decoder: super::ProtoByteDecoder<T, super::ByteDelimitedDecoder<T, bool, S>>,
585    }
586
587    impl<
588        T: FromKiss + Clone + Send + 'static,
589        S: Default + Send + Serialize + for<'de> Deserialize<'de> + 'static,
590        const GFEND: u8,
591        const GFESC: u8,
592        const GTFEND: u8,
593        const GTFESC: u8,
594    > ProtoKissModel<T, S, GFEND, GFESC, GTFEND, GTFESC>
595    {
596        /// Creates new KISS decoder.
597        pub fn new<F>(decode: F) -> Self
598        where
599            F: Fn(&mut S, &[u8]) -> T + Send + 'static,
600        {
601            let decoder = super::ProtoByteDecoder::new(
602                super::ByteDelimitedDecoder::new(GFEND, GFEND),
603                super::ByteDelimitedDecoderEnv::<T, bool, S>::with_transform(
604                    |is_esc: &mut bool, previous: &[u8], byte: u8| {
605                        if *is_esc {
606                            *is_esc = false;
607                            // Matching is not possible here because of generics,
608                            // see `rustc --explain E015`.
609                            if byte == GTFEND {
610                                Ok(super::Transformed::One(GFEND))
611                            } else if byte == GTFESC {
612                                Ok(super::Transformed::One(GFESC))
613                            } else {
614                                Err(T::abort_variant(previous, byte))
615                            }
616                        } else if byte == GFESC {
617                            *is_esc = true;
618                            Ok(super::Transformed::None)
619                        } else {
620                            Ok(super::Transformed::One(byte))
621                        }
622                    },
623                    decode,
624                ),
625            );
626            Self {
627                data_out: decoder.data_out.clone(),
628                decoder,
629            }
630        }
631    }
632
633    /// KISS decoder model.
634    pub type KissModel<T, S> = super::ByteDecoderModel<T, super::ByteDelimitedDecoder<T, bool, S>>;
635
636    impl<
637        T: FromKiss + Clone + Send + 'static,
638        S: Default + Send + Serialize + for<'de> Deserialize<'de> + 'static,
639        const GFEND: u8,
640        const GFESC: u8,
641        const GTFEND: u8,
642        const GTFESC: u8,
643    > ProtoModel for ProtoKissModel<T, S, GFEND, GFESC, GTFEND, GTFESC>
644    {
645        type Model = KissModel<T, S>;
646
647        fn build(
648            self,
649            _: &mut BuildContext<Self>,
650        ) -> (Self::Model, <Self::Model as model::Model>::Env) {
651            self.decoder.build()
652        }
653    }
654
655    impl<
656        T: FromKiss + Clone + Send + 'static,
657        S: Default + Send + Serialize + for<'de> Deserialize<'de> + 'static,
658        const GFEND: u8,
659        const GFESC: u8,
660        const GTFEND: u8,
661        const GTFESC: u8,
662    > fmt::Debug for ProtoKissModel<T, S, GFEND, GFESC, GTFEND, GTFESC>
663    {
664        fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
665            f.debug_struct("ProtoByteDecoder").finish_non_exhaustive()
666        }
667    }
668}
nexosim_byte_utils/decode.rs

nexosim_byte_utils/
decode.rs
