Tedge flows publish enabled/updated/removed status

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

Author: didier-wenzek

crates/extensions/tedge_flows/src/actor.rs

@@ -10,6 +10,7 @@ use crate::Tick;
 use async_trait::async_trait;
 use camino::Utf8PathBuf;
 use futures::FutureExt;
+use serde_json::json;
 use std::cmp::min;
 use std::collections::HashSet;
 use std::time::Duration;
@@ -21,10 +22,13 @@ use tedge_actors::Sender;
 use tedge_actors::SimpleMessageBox;
 use tedge_file_system_ext::FsWatchEvent;
 use tedge_mqtt_ext::MqttMessage;
+use tedge_mqtt_ext::QoS;
 use tedge_mqtt_ext::SubscriptionDiff;
+use tedge_mqtt_ext::Topic;
 use tedge_mqtt_ext::TopicFilter;
 use tedge_watch_ext::WatchEvent;
 use tedge_watch_ext::WatchRequest;
+use time::OffsetDateTime;
 use tokio::io::AsyncWriteExt;
 use tokio::time::sleep_until;
 use tokio::time::Instant;
@@ -52,6 +56,7 @@ impl Actor for FlowsMapper {
 
     async fn run(mut self) -> Result<(), RuntimeError> {
         self.send_updated_subscriptions().await?;
+        self.notify_flows_status().await?;
 
         while let Some(message) = self.next_message().await {
             match message {
@@ -73,17 +78,19 @@ impl Actor for FlowsMapper {
                     if matches!(path.extension(), Some("js" | "ts" | "mjs")) {
                         self.processor.reload_script(path).await;
                     } else if path.extension() == Some("toml") {
-                        self.processor.add_flow(path).await;
+                        self.processor.add_flow(path.clone()).await;
                         self.send_updated_subscriptions().await?;
+                        self.update_flow_status(path.as_str()).await?;
                     }
                 }
                 InputMessage::FsWatchEvent(FsWatchEvent::FileCreated(path)) => {
                     let Ok(path) = Utf8PathBuf::try_from(path) else {
                         continue;
                     };
                     if matches!(path.extension(), Some("toml")) {
-                        self.processor.add_flow(path).await;
+                        self.processor.add_flow(path.clone()).await;
                         self.send_updated_subscriptions().await?;
+                        self.update_flow_status(path.as_str()).await?;
                     }
                 }
                 InputMessage::FsWatchEvent(FsWatchEvent::FileDeleted(path)) => {
@@ -93,8 +100,9 @@ impl Actor for FlowsMapper {
                     if matches!(path.extension(), Some("js" | "ts" | "mjs")) {
                         self.processor.remove_script(path).await;
                     } else if path.extension() == Some("toml") {
-                        self.processor.remove_flow(path).await;
+                        self.processor.remove_flow(path.clone()).await;
                         self.send_updated_subscriptions().await?;
+                        self.update_flow_status(path.as_str()).await?;
                     }
                 }
                 _ => continue,
@@ -163,6 +171,38 @@ impl FlowsMapper {
         watch_requests
     }
 
+    async fn notify_flows_status(&mut self) -> Result<(), RuntimeError> {
+        let status = "enabled";
+        let now = OffsetDateTime::now_utc();
+        for flow in self.processor.flows.keys() {
+            let status = Self::flow_status(flow, status, &now);
+            self.mqtt_sender.send(status).await?;
+        }
+        Ok(())
+    }
+
+    async fn update_flow_status(&mut self, flow: &str) -> Result<(), RuntimeError> {
+        let now = OffsetDateTime::now_utc();
+        let status = if self.processor.flows.contains_key(flow) {
+            "updated"
+        } else {
+            "removed"
+        };
+        let status = Self::flow_status(flow, status, &now);
+        self.mqtt_sender.send(status).await?;
+        Ok(())
+    }
+
+    fn flow_status(flow: &str, status: &str, time: &OffsetDateTime) -> MqttMessage {
+        let topic = Topic::new_unchecked("te/device/main/service/tedge-flows/status/flows");
+        let payload = json!({
+            "flow": flow,
+            "status": status,
+            "time": time.unix_timestamp(),
+        });
+        MqttMessage::new(&topic, payload.to_string()).with_qos(QoS::AtLeastOnce)
+    }
+
     async fn on_source_poll(&mut self) -> Result<(), RuntimeError> {
         let now = Instant::now();
         let timestamp = DateTime::now();

tests/RobotFramework/tests/tedge_flows/tedge_flows.robot

@@ -120,7 +120,13 @@ Consuming messages from the tail of file
     [Teardown]    Uninstall Flow    tail-named-pipe.toml
 
 Consuming messages from a file, periodically
+    ${start}    Get Unix Timestamp
     Install Flow    read-file-periodically.toml
+    Should Have MQTT Messages
+    ...    topic=te/device/main/service/tedge-flows/status/flows
+    ...    minimum=1
+    ...    message_contains=read-file-periodically
+    ...    date_from=${start}
     Execute Command    echo hello >/tmp/file.input
     Execute Command    tedge mqtt sub test/file/input --duration 1s | grep hello
     Execute Command    echo world >/tmp/file.input
@@ -133,16 +139,27 @@ Consuming messages from a file, periodically
     [Teardown]    Uninstall Flow    read-file-periodically.toml
 
 Appending messages to a file
+    ${start}    Get Unix Timestamp
     Install Flow    append-to-file.toml
+    Should Have MQTT Messages
+    ...    topic=te/device/main/service/tedge-flows/status/flows
+    ...    minimum=1
+    ...    message_contains=append-to-file
+    ...    date_from=${start}
     Execute Command    for i in $(seq 3); do tedge mqtt pub seq/events "$i"; done
     Execute Command    grep '\\[seq/events\\] 1' /tmp/events.log
     Execute Command    grep '\\[seq/events\\] 2' /tmp/events.log
     Execute Command    grep '\\[seq/events\\] 3' /tmp/events.log
     [Teardown]    Uninstall Flow    append-to-file.toml
 
 Publishing transformation errors
-    Install Flow    publish-js-errors.toml
     ${start}    Get Unix Timestamp
+    Install Flow    publish-js-errors.toml
+    Should Have MQTT Messages
+    ...    topic=te/device/main/service/tedge-flows/status/flows
+    ...    minimum=1
+    ...    message_contains=publish-js-errors
+    ...    date_from=${start}
     Execute Command    tedge mqtt pub collectd/foo 12345:6789
     Should Have MQTT Messages
     ...    topic=test/errors