Allow nested parsers on MessageProducer level (v2)

application/apps/indexer/Cargo.toml

@@ -22,7 +22,9 @@ thiserror = "1.0"
 lazy_static = "1.4"
 tokio = { version = "1", features = ["full"] }
 tokio-stream = "0.1"
-dlt-core = "0.14"
+# dlt-core = "0.14"
+# TODO https://github.com/esrlabs/dlt-core/pull/24
+dlt-core = { git = "https://github.com/kruss/dlt-core.git", branch = "dlt_network_traces" }
 crossbeam-channel = "0.5"
 futures = "0.3"
 tokio-util = "0.7"

application/apps/indexer/parsers/Cargo.toml

@@ -18,7 +18,9 @@ rand.workspace = true
 # someip-messages = { path = "../../../../../someip"}
 someip-messages = { git = "https://github.com/esrlabs/someip" }
 # someip-payload = { path = "../../../../../someip-payload" }
-someip-payload = { git = "https://github.com/esrlabs/someip-payload" }
+# someip-payload = { git = "https://github.com/esrlabs/someip-payload" }
+# TODO
+someip-payload = { git = "https://github.com/kruss/someip-payload.git", branch = "robustness" }
 
 [dev-dependencies]
 stringreader = "0.1.1"

application/apps/indexer/parsers/src/dlt/fmt.rs

@@ -28,6 +28,7 @@ use log::trace;
 use serde::ser::{Serialize, SerializeStruct, Serializer};
 
 use std::{
+    collections::HashMap,
     fmt::{self, Formatter},
     str,
 };
@@ -193,11 +194,47 @@ impl From<Option<&String>> for FormatOptions {
     }
 }
 
+#[derive(Hash, Clone, Copy, PartialEq, Eq)]
+pub enum UnresolvedDataType {
+    SomeIpPayload,
+}
+
+pub struct UnresolvedData {
+    data: Vec<UnresolvedDataType>,
+    requested: Option<UnresolvedDataType>,
+}
+
+impl UnresolvedData {
+    pub fn next_unresolved(&mut self) -> Option<UnresolvedDataType> {
+        if self.data.is_empty() {
+            self.requested = None;
+            None
+        } else {
+            let requested = self.data.remove(0);
+            self.requested = Some(requested);
+            Some(requested)
+        }
+    }
+    pub fn requested(&self) -> Option<UnresolvedDataType> {
+        self.requested
+    }
+}
+
+impl Default for UnresolvedData {
+    fn default() -> Self {
+        Self {
+            data: vec![UnresolvedDataType::SomeIpPayload],
+            requested: None,
+        }
+    }
+}
 /// A dlt message that can be formatted with optional FIBEX data support
 pub struct FormattableMessage<'a> {
     pub message: Message,
     pub fibex_metadata: Option<&'a FibexMetadata>,
     pub options: Option<&'a FormatOptions>,
+    pub unresolved: UnresolvedData,
+    pub resolved: HashMap<UnresolvedDataType, Result<String, String>>,
 }
 
 impl<'a> Serialize for FormattableMessage<'a> {
@@ -281,6 +318,17 @@ impl<'a> Serialize for FormattableMessage<'a> {
                     None => state.serialize_field("payload", "[Unknown CtrlCommand]")?,
                 }
             }
+            PayloadContent::NetworkTrace(slices) => {
+                state.serialize_field("app-id", &ext_header_app_id)?;
+                state.serialize_field("context-id", &ext_header_context_id)?;
+                state.serialize_field("message-type", &ext_header_msg_type)?;
+                let arg_string = slices
+                    .iter()
+                    .map(|slice| format!("{:02X?}", slice))
+                    .collect::<Vec<String>>()
+                    .join("|");
+                state.serialize_field("payload", &arg_string)?;
+            }
         }
         state.end()
     }
@@ -292,6 +340,8 @@ impl<'a> From<Message> for FormattableMessage<'a> {
             message,
             fibex_metadata: None,
             options: None,
+            unresolved: UnresolvedData::default(),
+            resolved: HashMap::new(),
         }
     }
 }
@@ -320,6 +370,27 @@ impl<'a> PrintableMessage<'a> {
 }
 
 impl<'a> FormattableMessage<'a> {
+    pub fn get_someip_payload(&self) -> Option<&[u8]> {
+        if let PayloadContent::NetworkTrace(slices) = &self.message.payload {
+            if self
+                .message
+                .extended_header
+                .as_ref()
+                .is_some_and(|ext_header| {
+                    matches!(
+                        ext_header.message_type,
+                        MessageType::NetworkTrace(NetworkTraceType::Ipc)
+                            | MessageType::NetworkTrace(NetworkTraceType::Someip)
+                    )
+                })
+            {
+                if let Some(slice) = slices.get(1) {
+                    return Some(slice);
+                }
+            }
+        }
+        None
+    }
     pub fn printable_parts<'b>(
         &'b self,
         ext_h_app_id: &'b str,
@@ -386,6 +457,19 @@ impl<'a> FormattableMessage<'a> {
                     payload_string,
                 ))
             }
+            PayloadContent::NetworkTrace(slices) => {
+                let payload_string = slices
+                    .iter()
+                    .map(|slice| format!("{:02X?}", slice))
+                    .collect::<Vec<String>>()
+                    .join("|");
+                Ok(PrintableMessage::new(
+                    ext_h_app_id,
+                    eh_ctx_id,
+                    ext_h_msg_type,
+                    payload_string,
+                ))
+            }
         }
     }
 
@@ -559,6 +643,24 @@ impl<'a> fmt::Display for FormattableMessage<'a> {
                     None => write!(f, "[Unknown CtrlCommand]"),
                 }
             }
+            PayloadContent::NetworkTrace(slices) => {
+                self.write_app_id_context_id_and_message_type(f)?;
+                if let Some(resolved) = self.resolved.get(&UnresolvedDataType::SomeIpPayload) {
+                    match resolved {
+                        Ok(msg) => write!(f, "SOME/IP: {}", msg.replace("", " | ")),
+                        Err(err) => {
+                            write!(f, "SOME/IP PARSING ERROR: {err}; ORIGIN: ")?;
+                            slices.iter().try_for_each(|slice| {
+                                write!(f, "{}{:02X?}", DLT_ARGUMENT_SENTINAL, slice)
+                            })
+                        }
+                    }
+                } else {
+                    slices
+                        .iter()
+                        .try_for_each(|slice| write!(f, "{}{:02X?}", DLT_ARGUMENT_SENTINAL, slice))
+                }
+            }
         }
     }
 }

application/apps/indexer/parsers/src/dlt/mod.rs

@@ -1,19 +1,22 @@
 pub mod attachment;
 pub mod fmt;
 
-use crate::{dlt::fmt::FormattableMessage, Error, LogMessage, ParseYield, Parser};
+use crate::{
+    dlt::fmt::FormattableMessage, Error, LogMessage, ParseYield, Parser, ParserInstanceAlias,
+};
 use byteorder::{BigEndian, WriteBytesExt};
+use dlt_core::{
+    dlt,
+    parse::{dlt_consume_msg, dlt_message},
+};
 pub use dlt_core::{
     dlt::LogLevel,
     fibex::{gather_fibex_data, FibexConfig, FibexMetadata},
     filtering::{DltFilterConfig, ProcessedDltFilterConfig},
 };
-use dlt_core::{
-    dlt::{self},
-    parse::{dlt_consume_msg, dlt_message},
-};
+use fmt::{UnresolvedData, UnresolvedDataType};
 use serde::Serialize;
-use std::{io::Write, ops::Range};
+use std::{collections::HashMap, io::Write, ops::Range};
 
 use self::{attachment::FtScanner, fmt::FormatOptions};
 
@@ -24,6 +27,26 @@ impl LogMessage for FormattableMessage<'_> {
         writer.write_all(&bytes)?;
         Ok(len)
     }
+    fn next_unresolved(&mut self) -> Option<(ParserInstanceAlias, &[u8])> {
+        if let Some(next) = self.unresolved.next_unresolved() {
+            match next {
+                UnresolvedDataType::SomeIpPayload => {
+                    if let Some(payload) = self.get_someip_payload() {
+                        return Some((ParserInstanceAlias::SomeIp, payload));
+                    }
+                }
+            }
+            None
+        } else {
+            None
+        }
+    }
+    fn resolve(&mut self, result: Option<Result<String, String>>) {
+        let (Some(reqested), Some(parsed)) = (self.unresolved.requested(), result) else {
+            return;
+        };
+        self.resolved.insert(reqested, parsed);
+    }
 }
 
 #[derive(Debug, Serialize)]
@@ -143,6 +166,8 @@ impl<'m> Parser<FormattableMessage<'m>> for DltParser<'m> {
                     message: msg_with_storage_header,
                     fibex_metadata: self.fibex_metadata,
                     options: self.fmt_options,
+                    unresolved: UnresolvedData::default(),
+                    resolved: HashMap::new(),
                 };
                 self.offset += input.len() - rest.len();
                 Ok((

application/apps/indexer/parsers/src/lib.rs

@@ -8,6 +8,16 @@ use thiserror::Error;
 
 extern crate log;
 
+#[derive(Debug, Clone, Copy)]
+pub enum ParserInstanceAlias {
+    SomeIp,
+}
+
+#[allow(clippy::large_enum_variant)]
+pub enum ParserInstance {
+    SomeIp(someip::SomeipParser),
+}
+
 #[derive(Error, Debug)]
 pub enum Error {
     #[error("Parse error: {0}")]
@@ -82,6 +92,10 @@ pub trait LogMessage: Display + Serialize {
     /// Serializes a message directly into a Writer
     /// returns the size of the serialized message
     fn to_writer<W: Write>(&self, writer: &mut W) -> Result<usize, std::io::Error>;
+    fn next_unresolved(&mut self) -> Option<(ParserInstanceAlias, &[u8])> {
+        None
+    }
+    fn resolve(&mut self, _result: Option<Result<String, String>>) {}
 }
 
 #[derive(Debug)]

application/apps/indexer/session/Cargo.toml

@@ -8,7 +8,9 @@ edition = "2021"
 blake3 = "1.3"
 crossbeam-channel.workspace = true
 dirs.workspace = true
-dlt-core.workspace = true
+# dlt-core.workspace = true
+# TODO https://github.com/esrlabs/dlt-core/pull/24
+dlt-core = { git = "https://github.com/kruss/dlt-core.git", branch = "dlt_network_traces", features=["statistics"] }
 envvars = "0.1"
 file-tools = { path = "../addons/file-tools" }
 futures.workspace = true