Distinguish two message sources: MQTT vs DB

Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>

Author: didier-wenzek

crates/core/tedge/src/cli/flows/test.rs

@@ -75,7 +75,7 @@ impl TestCommand {
             }
         };
         processor
-            .on_message(timestamp, &message)
+            .on_message(MessageSource::MQTT, timestamp, &message)
             .await
             .into_iter()
             .for_each(|msg| self.print_messages(msg))

crates/extensions/tedge_flows/src/actor.rs

@@ -1,6 +1,7 @@
 use crate::flow::DateTime;
 use crate::flow::FlowOutput;
 use crate::flow::Message;
+use crate::flow::MessageSource;
 use crate::runtime::MessageProcessor;
 use crate::InputMessage;
 use crate::OutputMessage;
@@ -22,6 +23,7 @@ use tedge_mqtt_ext::TopicFilter;
 use tokio::time::sleep_until;
 use tokio::time::Instant;
 use tracing::error;
+use tracing::info;
 
 pub const STATS_DUMP_INTERVAL: Duration = Duration::from_secs(300);
 
@@ -42,13 +44,17 @@ impl Actor for FlowsMapper {
         while let Some(message) = self.next_message().await {
             match message {
                 InputMessage::Tick(_) => {
-                    self.on_interval().await?;
-
                     let drained_messages = self.drain_db().await?;
-                    self.filter_all(drained_messages).await?;
+                    self.on_messages(MessageSource::MeaDB, drained_messages)
+                        .await?;
+
+                    self.on_interval().await?;
                 }
                 InputMessage::MqttMessage(message) => match Message::try_from(message) {
-                    Ok(message) => self.on_message(message).await?,
+                    Ok(message) => {
+                        self.on_message(MessageSource::MQTT, DateTime::now(), message)
+                            .await?
+                    }
                     Err(err) => {
                         error!(target: "flows", "Cannot process message: {err}");
                     }
@@ -125,24 +131,16 @@ impl FlowsMapper {
         diff
     }
 
-    async fn on_message(&mut self, message: Message) -> Result<(), RuntimeError> {
-        let timestamp = DateTime::now();
-        for (flow_id, flow_messages) in self.processor.on_message(timestamp, &message).await {
+    async fn on_message(
+        &mut self,
+        source: MessageSource,
+        timestamp: DateTime,
+        message: Message,
+    ) -> Result<(), RuntimeError> {
+        for (flow_id, flow_messages) in self.processor.on_message(source, timestamp, &message).await
+        {
             match flow_messages {
-                Ok(messages) => {
-                    for message in messages {
-                        match MqttMessage::try_from(message) {
-                            Ok(message) => {
-                                self.messages
-                                    .send(OutputMessage::MqttMessage(message))
-                                    .await?
-                            }
-                            Err(err) => {
-                                error!(target: "flows", "{flow_id}: cannot send transformed message: {err}")
-                            }
-                        }
-                    }
-                }
+                Ok(messages) => self.publish_messages(flow_id, timestamp, messages).await?,
                 Err(err) => {
                     error!(target: "flows", "{flow_id}: {err}");
                 }
@@ -152,26 +150,14 @@ impl FlowsMapper {
         Ok(())
     }
 
-    async fn filter_all(&mut self, messages: Vec<(DateTime, Message)>) -> Result<(), RuntimeError> {
+    async fn on_messages(
+        &mut self,
+        source: MessageSource,
+        messages: Vec<(DateTime, Message)>,
+    ) -> Result<(), RuntimeError> {
         for (timestamp, message) in messages {
-            self.filter(timestamp, message).await?
-        }
-        Ok(())
-    }
-
-    async fn filter(&mut self, timestamp: DateTime, message: Message) -> Result<(), RuntimeError> {
-        for (flow_id, flow_messages) in self.processor.process(timestamp, &message).await {
-            match flow_messages {
-                Ok(messages) => {
-                    self.publish_messages(flow_id.clone(), timestamp, messages)
-                        .await?;
-                }
-                Err(err) => {
-                    error!(target: "flows", "{flow_id}: {err}");
-                }
-            }
+            self.on_message(source, timestamp, message).await?
         }
-
         Ok(())
     }
 
@@ -225,6 +211,7 @@ impl FlowsMapper {
                 }
                 FlowOutput::MeaDB { output_series } => {
                     for message in messages {
+                        info!(target: "flows", "store {output_series} @{}.{} [{}]", timestamp.seconds, timestamp.nanoseconds, message.topic);
                         if let Err(err) = self
                             .processor
                             .database
@@ -246,6 +233,9 @@ impl FlowsMapper {
         for (flow_id, flow_messages) in self.processor.drain_db(timestamp).await {
             match flow_messages {
                 Ok(flow_messages) => {
+                    for (t, m) in flow_messages.iter() {
+                        info!(target: "flows", "drained: @{}.{} [{}]", t.seconds, t.nanoseconds, m.topic);
+                    }
                     messages.extend(flow_messages);
                 }
                 Err(err) => {

crates/extensions/tedge_flows/src/bin/db_dump.rs

@@ -0,0 +1,34 @@
+use anyhow::Context;
+use tedge_flows::flow::DateTime;
+use tedge_flows::flow::Message;
+
+const DB_PATH: &str = "/etc/tedge/tedge-gen.db";
+
+type Record = (DateTime, Message);
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    let mut args = std::env::args();
+    let series_name = args.nth(1).unwrap_or("latest-data-points".to_string());
+    println!("Reading series name: {series_name}");
+
+    let mut db = tedge_flows::MeaDB::open(DB_PATH)
+        .await
+        .with_context(|| format!("Failed to open DB at path={DB_PATH}"))?;
+
+    let items = db
+        .query_all(&series_name)
+        .await
+        .context("Failed to query for all items")?;
+
+    items.iter().for_each(print_record);
+
+    Ok(())
+}
+
+fn print_record(record: &Record) {
+    let time = record.0.seconds;
+    let topic = &record.1.topic;
+    let payload = std::str::from_utf8(&record.1.payload).unwrap_or("<binary payload>");
+    println!("[{time}]\t{topic}\t{payload}")
+}

crates/extensions/tedge_flows/src/config.rs

@@ -122,12 +122,16 @@ impl FlowConfig {
             steps.push(step);
         }
         let output = self.output.try_into()?;
-        Ok(Flow {
+        let mut flow = Flow {
             input,
             steps,
             source,
             output,
-        })
+            next_drain: None,
+            last_drain: None,
+        };
+        flow.init_next_drain();
+        Ok(flow)
     }
 }
 

crates/extensions/tedge_flows/src/flow.rs

@@ -13,6 +13,7 @@ use tokio::time::Instant;
 use tracing::warn;
 
 /// A chain of transformation of MQTT messages
+#[derive(Debug)]
 pub struct Flow {
     /// The source topics
     pub input: FlowInput,
@@ -24,14 +25,22 @@ pub struct Flow {
 
     /// Target of the transformed messages
     pub output: FlowOutput,
+
+    /// Next time to drain database for MeaDB inputs (for deadline-based wakeup)
+    pub next_drain: Option<tokio::time::Instant>,
+
+    /// Last time database was drained (for frequency checking)
+    pub last_drain: Option<DateTime>,
 }
 
 /// A message transformation step
+#[derive(Debug)]
 pub struct FlowStep {
     pub script: JsScript,
     pub config_topics: TopicFilter,
 }
 
+#[derive(Debug)]
 pub enum FlowInput {
     MQTT {
         topics: TopicFilter,
@@ -49,6 +58,12 @@ pub enum FlowOutput {
     MeaDB { output_series: String },
 }
 
+#[derive(Copy, Clone, Eq, PartialEq)]
+pub enum MessageSource {
+    MQTT,
+    MeaDB,
+}
+
 #[derive(
     Copy, Clone, Debug, serde::Deserialize, serde::Serialize, Eq, PartialEq, Ord, PartialOrd,
 )]
@@ -77,8 +92,57 @@ pub enum FlowError {
 }
 
 impl Flow {
+    pub fn accept(&self, source: MessageSource, message_topic: &str) -> bool {
+        match &self.input {
+            FlowInput::MQTT {
+                topics: input_topics,
+            } => source == MessageSource::MQTT && input_topics.accept_topic_name(message_topic),
+            FlowInput::MeaDB { .. } => source == MessageSource::MeaDB,
+        }
+    }
+
+    pub fn init_next_drain(&mut self) {
+        if let FlowInput::MeaDB { frequency, .. } = &self.input {
+            if !frequency.is_zero() {
+                self.next_drain = Some(tokio::time::Instant::now() + *frequency);
+            }
+        }
+    }
+
+    pub fn should_drain_at(&mut self, timestamp: DateTime) -> bool {
+        if let FlowInput::MeaDB { frequency, .. } = &self.input {
+            if frequency.is_zero() {
+                return false;
+            }
+
+            // Check if enough time has passed since last drain
+            match self.last_drain {
+                Some(last_drain) => {
+                    let elapsed_secs = timestamp.seconds.saturating_sub(last_drain.seconds);
+                    let frequency_secs = frequency.as_secs();
+                    if elapsed_secs >= frequency_secs {
+                        self.last_drain = Some(timestamp);
+                        // Also update the deadline for the actor loop
+                        self.next_drain = Some(tokio::time::Instant::now() + *frequency);
+                        true
+                    } else {
+                        false
+                    }
+                }
+                None => {
+                    // First drain
+                    self.last_drain = Some(timestamp);
+                    self.next_drain = Some(tokio::time::Instant::now() + *frequency);
+                    true
+                }
+            }
+        } else {
+            false
+        }
+    }
+
     pub fn topics(&self) -> TopicFilter {
-        let mut topics = self.input.topics().clone();
+        let mut topics = self.input.topics();
         for step in self.steps.iter() {
             topics.add_all(step.config_topics.clone())
         }
@@ -101,12 +165,13 @@ impl Flow {
     pub async fn on_message(
         &mut self,
         js_runtime: &JsRuntime,
+        source: MessageSource,
         stats: &mut Counter,
         timestamp: DateTime,
         message: &Message,
     ) -> Result<Vec<Message>, FlowError> {
         self.on_config_update(js_runtime, message).await?;
-        if !self.input.topics().accept_topic_name(&message.topic) {
+        if !self.accept(source, &message.topic) {
             return Ok(vec![]);
         }
 
@@ -203,14 +268,12 @@ impl FlowStep {
 }
 
 impl FlowInput {
-    pub fn topics(&self) -> &TopicFilter {
+    fn topics(&self) -> TopicFilter {
         match self {
-            FlowInput::MQTT { topics } => topics,
+            FlowInput::MQTT { topics } => topics.clone(),
             FlowInput::MeaDB { .. } => {
                 // MeaDB inputs don't subscribe to MQTT topics
-                // Return an empty topic filter
-                static EMPTY_TOPICS: std::sync::OnceLock<TopicFilter> = std::sync::OnceLock::new();
-                EMPTY_TOPICS.get_or_init(TopicFilter::empty)
+                TopicFilter::empty()
             }
         }
     }

crates/extensions/tedge_flows/src/lib.rs

@@ -32,6 +32,8 @@ use tedge_mqtt_ext::TopicFilter;
 use tokio::time::Instant;
 use tracing::error;
 
+pub use runtime::MeaDB;
+
 fan_in_message_type!(InputMessage[MqttMessage, FsWatchEvent, Tick]: Clone, Debug, Eq, PartialEq);
 fan_in_message_type!(OutputMessage[MqttMessage, SubscriptionDiff]: Clone, Debug, Eq, PartialEq);