Added robustness on reading DLT messages

Author: kruss

CHANGELOG.md

@@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.20.1] - 2025-05-09
+
+### Changed
+
+- Added robustness for DLT message reader and stream
+
 ## [0.20.0] - 2025-02-25
 
 ### Added

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "dlt-core"
-version = "0.20.0"
+version = "0.20.1"
 authors = ["esrlabs.com"]
 edition = "2021"
 description = """

README.md

@@ -69,7 +69,10 @@ The following example can be run with `cargo run --example file_parser --release
 
 <!-- example start -->
 ```rust
-use dlt_core::read::{read_message, DltMessageReader};
+use dlt_core::{
+    parse::DltParseError,
+    read::{read_message, DltMessageReader},
+};
 use std::{env, fs, fs::File, path::PathBuf, time::Instant};
 
 fn main() {
@@ -81,8 +84,21 @@ fn main() {
     let mut dlt_reader = DltMessageReader::new(dlt_file, true);
     let mut message_count = 0usize;
     let start = Instant::now();
-    while let Some(_message) = read_message(&mut dlt_reader, None).expect("read dlt message") {
-        message_count += 1;
+    loop {
+        match read_message(&mut dlt_reader, None) {
+            Ok(Some(_)) => {
+                message_count += 1;
+            }
+            Ok(None) => {
+                break;
+            }
+            Err(error) => match error {
+                DltParseError::ParsingHickup(_) => {
+                    continue;
+                }
+                _ => panic!("{}", error),
+            },
+        }
     }
     // print some stats
     let duration_in_s = start.elapsed().as_millis() as f64 / 1000.0;

examples/file_parser.rs

@@ -1,4 +1,7 @@
-use dlt_core::read::{read_message, DltMessageReader};
+use dlt_core::{
+    parse::DltParseError,
+    read::{read_message, DltMessageReader},
+};
 use std::{env, fs, fs::File, path::PathBuf, time::Instant};
 
 fn main() {
@@ -10,8 +13,21 @@ fn main() {
     let mut dlt_reader = DltMessageReader::new(dlt_file, true);
     let mut message_count = 0usize;
     let start = Instant::now();
-    while let Some(_message) = read_message(&mut dlt_reader, None).expect("read dlt message") {
-        message_count += 1;
+    loop {
+        match read_message(&mut dlt_reader, None) {
+            Ok(Some(_)) => {
+                message_count += 1;
+            }
+            Ok(None) => {
+                break;
+            }
+            Err(error) => match error {
+                DltParseError::ParsingHickup(_) => {
+                    continue;
+                }
+                _ => panic!("{}", error),
+            },
+        }
     }
     // print some stats
     let duration_in_s = start.elapsed().as_millis() as f64 / 1000.0;

examples/file_streamer.rs

@@ -1,4 +1,7 @@
-use dlt_core::stream::{read_message, DltStreamReader};
+use dlt_core::{
+    parse::DltParseError,
+    stream::{read_message, DltStreamReader},
+};
 use std::{env, fs, path::PathBuf, time::Instant};
 use tokio::fs::File;
 use tokio_util::compat::TokioAsyncReadCompatExt;
@@ -13,11 +16,21 @@ async fn main() {
     let mut dlt_reader = DltStreamReader::new(dlt_file.compat(), true);
     let mut message_count = 0usize;
     let start = Instant::now();
-    while let Some(_message) = read_message(&mut dlt_reader, None)
-        .await
-        .expect("read dlt message")
-    {
-        message_count += 1;
+    loop {
+        match read_message(&mut dlt_reader, None).await {
+            Ok(Some(_)) => {
+                message_count += 1;
+            }
+            Ok(None) => {
+                break;
+            }
+            Err(error) => match error {
+                DltParseError::ParsingHickup(_) => {
+                    continue;
+                }
+                _ => panic!("{}", error),
+            },
+        }
     }
     // print some stats
     let duration_in_s = start.elapsed().as_millis() as f64 / 1000.0;

src/dlt.rs

@@ -901,7 +901,8 @@ impl TryFrom<u32> for TypeInfo {
 ///
 /// * phy_v is what we received in the dlt message
 /// * log_v is the real value
-///     example: the degree celcius is transmitted,
+///
+/// Example: the degree celcius is transmitted,
 ///     quantization = 0.01, offset = -50
 ///     now the transmitted value phy_v = (log_v - offset)/quantization = 7785
 ///

src/read.rs

@@ -16,7 +16,7 @@
 use crate::{
     dlt::{HEADER_MIN_LENGTH, STORAGE_HEADER_LENGTH},
     filtering::ProcessedDltFilterConfig,
-    parse::{dlt_message, parse_length, DltParseError, ParsedMessage},
+    parse::{dlt_message, parse_length, DltParseError, ParsedMessage, DLT_PATTERN},
 };
 use std::io::{BufReader, Read};
 
@@ -69,7 +69,7 @@ impl<S: Read> DltMessageReader<S> {
         source: S,
         with_storage_header: bool,
     ) -> Self {
-        debug_assert!(buffer_capacity >= message_max_len);
+        assert!(buffer_capacity >= message_max_len);
 
         DltMessageReader {
             source: BufReader::with_capacity(buffer_capacity, source),
@@ -81,30 +81,51 @@ impl<S: Read> DltMessageReader<S> {
     /// Read the next message slice from the source,
     /// or return an empty slice if no more message could be read.
     pub fn next_message_slice(&mut self) -> Result<&[u8], DltParseError> {
-        let storage_len = if self.with_storage_header {
-            STORAGE_HEADER_LENGTH as usize
-        } else {
-            0
-        };
-        let header_len = storage_len + HEADER_MIN_LENGTH as usize;
-        debug_assert!(header_len <= self.buffer.len());
-
-        if self
-            .source
-            .read_exact(&mut self.buffer[..header_len])
-            .is_err()
-        {
-            return Ok(&[]);
-        }
+        loop {
+            let storage_len = if self.with_storage_header {
+                let storage_len = STORAGE_HEADER_LENGTH as usize;
+
+                loop {
+                    if self
+                        .source
+                        .read_exact(&mut self.buffer[..storage_len])
+                        .is_err()
+                    {
+                        return Ok(&[]);
+                    }
+
+                    if &self.buffer[..DLT_PATTERN.len()] == DLT_PATTERN {
+                        break;
+                    }
+                }
+
+                storage_len
+            } else {
+                0
+            };
+
+            let header_len = storage_len + HEADER_MIN_LENGTH as usize;
+
+            if self
+                .source
+                .read_exact(&mut self.buffer[storage_len..header_len])
+                .is_err()
+            {
+                return Ok(&[]);
+            }
+
+            let (_, message_len) = parse_length(&self.buffer[storage_len..header_len])?;
 
-        let (_, message_len) = parse_length(&self.buffer[storage_len..header_len])?;
-        let total_len = storage_len + message_len as usize;
-        debug_assert!(total_len <= self.buffer.len());
+            let total_len = storage_len + message_len as usize;
+            if total_len < header_len {
+                continue;
+            }
 
-        self.source
-            .read_exact(&mut self.buffer[header_len..total_len])?;
+            self.source
+                .read_exact(&mut self.buffer[header_len..total_len])?;
 
-        Ok(&self.buffer[..total_len])
+            return Ok(&self.buffer[..total_len]);
+        }
     }
 
     /// Answer if message slices contain a `StorageHeader´.
@@ -163,15 +184,44 @@ mod tests {
                 assert_eq!(bytes, message.as_bytes());
             }
 
-            assert_eq!(None, read_message(&mut reader, None).expect("message"))
+            assert_eq!(None, read_message(&mut reader, None).expect("message"));
         }
     }
 
+    #[test]
+    fn test_read_message_robustness() {
+        #[rustfmt::skip]
+        let bytes = [
+            [
+                // --------------- storage header with invalid dlt-pattern
+                0xFF, 0x4C, 0x54, 0x01, 0x2B, 0x2C, 0xC9, 0x4D, 
+                0x7A, 0xE8, 0x01, 0x00, 0x45, 0x43, 0x55, 0x00, 
+            ]
+            .to_vec(),
+            [
+                // --------------- storage header
+                0x44, 0x4C, 0x54, 0x01, 0x2B, 0x2C, 0xC9, 0x4D, 
+                0x7A, 0xE8, 0x01, 0x00, 0x45, 0x43, 0x55, 0x00, 
+                // --------------- standard header with invalid length
+                0x21, 0x0A, 0x00, 0x00,
+            ]
+            .to_vec(),
+            DLT_MESSAGE_WITH_STORAGE_HEADER.to_vec(),
+        ]
+        .concat();
+
+        let mut reader = DltMessageReader::new(bytes.as_slice(), true);
+
+        assert!(read_message(&mut reader, None).expect("message").is_some());
+        assert!(read_message(&mut reader, None).expect("message").is_none());
+    }
+
     proptest! {
         #[test]
         fn test_read_messages_proptest(messages in messages_strat(10)) {
             test_read_messages(messages, false);
         }
+
         #[test]
         fn test_read_messages_with_storage_header_proptest(messages in messages_with_storage_header_strat(10)) {
             test_read_messages(messages, true);