Add support form parsing dlt-messages from streams

CHANGELOG.md

@@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.20.0] - 2025-02-11
+
+### Added
+
+- Support parsing of DLT messages from streams
+
 ## [0.19.2] - 2025-02-06
 
 ### Changed

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "dlt-core"
-version = "0.19.2"
+version = "0.20.0"
 authors = ["esrlabs.com"]
 edition = "2021"
 description = """
@@ -21,13 +21,25 @@ rustc-hash = { version = "2.1", optional = true }
 serde = { version = "1.0", features = ["derive"], optional = true }
 serde_json = { version = "1.0", optional = true }
 thiserror = "1.0"
+futures = "0.3"
+tokio = { version = "1", features = ["full"], optional = true }
+tokio-stream = { version = "0.1", optional = true }
+tokio-util = { version = "0.7", optional = true }
 
 [features]
 default = []
 statistics = ["buf_redux", "rustc-hash"]
 fibex_parser = ["quick-xml"]
 debug_parser = []
-serde-support = ["serde", "serde_json"]
+serde-support = [
+    "serde",
+    "serde_json"
+]
+stream-support = [ 
+    "tokio", 
+    "tokio-stream", 
+    "tokio-util" 
+]
 
 [lints.rust]
 unexpected_cfgs = { level = "warn", check-cfg = ['cfg(tarpaulin_include)'] }

README.md

@@ -156,6 +156,8 @@ Below is the revised and improved English version of the documentation:
 
 * **`serde-support`**: Adds `Serialize` and `Deserialize` implementations (via `serde`) to all public types. This feature is useful if you need to encode or decode these types for transmission or storage.
 
+* **`stream-support`**: Provides API for parsing DLT messages from streams.
+
 ## Example users
 
 ### Fast DLT Log Viewing with chipmunk

src/dlt.rs

@@ -1708,6 +1708,8 @@ pub(crate) const WITH_ECU_ID_FLAG: u8 = 1 << 2;
 pub(crate) const WITH_SESSION_ID_FLAG: u8 = 1 << 3;
 pub(crate) const WITH_TIMESTAMP_FLAG: u8 = 1 << 4;
 pub(crate) const HEADER_MIN_LENGTH: u16 = 4;
+#[cfg(feature = "stream-support")]
+pub(crate) const HEADER_MAX_LENGTH: u16 = 16;
 
 // Verbose Mode
 

src/lib.rs

@@ -30,6 +30,9 @@ pub mod service_id;
 #[cfg(not(tarpaulin_include))]
 #[cfg(feature = "statistics")]
 pub mod statistics;
+#[cfg(not(tarpaulin_include))]
+#[cfg(feature = "stream-support")]
+pub mod stream;
 
 #[cfg(test)]
 pub mod proptest_strategies;

src/statistics.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//! # rapidly gather statistics info of a dlt source
+//! # Rapidly gather statistics info of a dlt source
 use crate::{
     dlt::{LogLevel, MessageType},
     parse::{

src/stream.rs

@@ -0,0 +1,57 @@
+// Copyright 2021 by Accenture ESR
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! # Provide API to parse dlt-messages from streams
+use crate::{
+    dlt::{HEADER_MAX_LENGTH, STORAGE_HEADER_LENGTH},
+    filtering::ProcessedDltFilterConfig,
+    parse::{dlt_message, dlt_standard_header, DltParseError, ParsedMessage},
+};
+use futures::{AsyncRead, AsyncReadExt};
+
+/// Parse a DLT-message from a stream.
+///
+/// Note: This is an adapter for [`parse::dlt_message`]
+pub async fn dlt_stream<S: AsyncRead + Unpin>(
+    stream: &mut S,
+    filter_config_opt: Option<&ProcessedDltFilterConfig>,
+    with_storage_header: bool,
+) -> Result<ParsedMessage, DltParseError> {
+    let storage_len = if with_storage_header {
+        STORAGE_HEADER_LENGTH as usize
+    } else {
+        0
+    };
+
+    let header_len = storage_len + HEADER_MAX_LENGTH as usize;
+    let mut header_buf = vec![0u8; header_len];
+    stream.read_exact(&mut header_buf).await?;
+
+    let (_, header) = dlt_standard_header(&header_buf[storage_len..])?;
+
+    let message_len = storage_len + header.overall_length() as usize;
+    let mut message_buf = vec![0u8; message_len];
+    message_buf[..header_len].copy_from_slice(&header_buf);
+
+    stream.read_exact(&mut message_buf[header_len..]).await?;
+    let (rest, message) = dlt_message(&message_buf, filter_config_opt, with_storage_header)?;
+
+    if !rest.is_empty() {
+        return Err(DltParseError::Unrecoverable(format!(
+            "Incomplete parse ({} bytes remaining)!",
+            rest.len()
+        )));
+    }
+    Ok(message)
+}

src/tests/mod.rs

@@ -18,3 +18,5 @@ mod dlt_parse_tests;
 mod fibex_tests;
 #[cfg(feature = "statistics")]
 mod statistics_tests;
+#[cfg(feature = "stream-support")]
+mod stream_tests;

src/tests/stream_tests.rs

@@ -0,0 +1,50 @@
+// Copyright 2021 by Accenture ESR
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+#[cfg(test)]
+mod tests {
+    use crate::parse::ParsedMessage;
+    use crate::stream::dlt_stream;
+    use futures::{stream, TryStreamExt};
+
+    #[tokio::test]
+    async fn test_dlt_stream() {
+        #[rustfmt::skip]
+        let bytes: Vec<u8> = vec![
+            0x44, 0x4C, 0x54, 0x01, 0x46, 0x93, 0x01, 0x5D, 0x79, 0x39, 0x0E, 0x00, 0x48, 0x46,
+            0x50, 0x50, 0x3D, 0x1E, 0x00, 0xA8, 0x48, 0x46, 0x50, 0x50, 0x00, 0x00, 0x02, 0x48,
+            0x00, 0x1C, 0x76, 0x49, 0x51, 0x08, 0x50, 0x61, 0x72, 0x61, 0x76, 0x63, 0x73, 0x6F,
+            0x00, 0x82, 0x00, 0x00, 0x1A, 0x00, 0x5B, 0x35, 0x38, 0x34, 0x3A, 0x20, 0x53, 0x6F,
+            0x6D, 0x65, 0x49, 0x70, 0x50, 0x6F, 0x73, 0x69, 0x78, 0x43, 0x6C, 0x69, 0x65, 0x6E,
+            0x74, 0x5D, 0x20, 0x00, 0x00, 0x82, 0x00, 0x00, 0x12, 0x00, 0x53, 0x65, 0x6E, 0x64,
+            0x53, 0x6F, 0x6D, 0x65, 0x49, 0x70, 0x4D, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x00,
+            0x00, 0x82, 0x00, 0x00, 0x02, 0x00, 0x3A, 0x00, 0x23, 0x00, 0x00, 0x00, 0x10, 0x01,
+            0x00, 0x00, 0x00, 0x82, 0x00, 0x00, 0x11, 0x00, 0x3A, 0x20, 0x69, 0x6E, 0x73, 0x74,
+            0x61, 0x6E, 0x63, 0x65, 0x5F, 0x69, 0x64, 0x20, 0x30, 0x78, 0x00, 0x42, 0x00, 0x01,
+            0x00, 0x01, 0x00, 0x00, 0x82, 0x00, 0x00, 0x17, 0x00, 0x20, 0x6D, 0x65, 0x6D, 0x6F,
+            0x72, 0x79, 0x20, 0x62, 0x75, 0x66, 0x66, 0x65, 0x72, 0x20, 0x6C, 0x65, 0x6E, 0x67,
+            0x74, 0x68, 0x20, 0x00, 0x44, 0x00, 0x00, 0x00, 0x14, 0x00, 0x00, 0x00, 0x00, 0x00,
+            0x00, 0x00,
+        ];
+
+        let stream = stream::iter([Ok(bytes.clone())]);
+        let mut reader = stream.into_async_read();
+
+        match dlt_stream(&mut reader, None, true).await {
+            Ok(ParsedMessage::Item(message)) => {
+                assert_eq!(bytes, message.as_bytes());
+            }
+            _ => panic!("could not parse message"),
+        }
+    }
+}