Merge pull request #3831 from didier-wenzek/fix/missing-reload-events

fix: missing reload events

Author: didier-wenzek

crates/extensions/tedge_flows/src/actor.rs

@@ -73,7 +73,7 @@ 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.reload_flow(path).await;
+                        self.processor.add_flow(path).await;
                         self.send_updated_subscriptions().await?;
                     }
                 }

crates/extensions/tedge_flows/src/runtime.rs

@@ -95,10 +95,18 @@ impl MessageProcessor {
     }
 
     fn deadlines(&self) -> impl Iterator<Item = tokio::time::Instant> + '_ {
-        self.flows
+        let script_deadlines = self
+            .flows
             .values()
             .flat_map(|flow| &flow.steps)
-            .filter_map(|step| step.script.next_execution)
+            .filter_map(|step| step.script.next_execution);
+
+        let source_deadlines = self
+            .flows
+            .values()
+            .filter_map(|flow| flow.input.next_deadline());
+
+        script_deadlines.chain(source_deadlines)
     }
 
     /// Get the next deadline for interval execution across all scripts
@@ -241,22 +249,15 @@ impl MessageProcessor {
 
     pub async fn add_flow(&mut self, path: Utf8PathBuf) {
         let flow_id = Self::flow_id(&path);
-        if !self.flows.contains_key(&flow_id) && self.load_flow(flow_id, path.clone()).await {
-            info!(target: "flows", "Loaded new flow {path}");
-        }
-    }
-
-    pub async fn reload_flow(&mut self, path: Utf8PathBuf) {
-        let flow_id = Self::flow_id(&path);
-        if self.flows.contains_key(&flow_id) && self.load_flow(flow_id, path.clone()).await {
-            info!(target: "flows", "Reloaded updated flow {path}");
+        if self.load_flow(flow_id, path.clone()).await {
+            info!(target: "flows", "Loading flow {path}");
         }
     }
 
     pub async fn remove_flow(&mut self, path: Utf8PathBuf) {
         let flow_id = Self::flow_id(&path);
         self.flows.remove(&flow_id);
-        info!(target: "flows", "Removed deleted flow {path}");
+        info!(target: "flows", "Removing flow {path}");
     }
 }
 

tests/RobotFramework/tests/tedge_flows/flows/append-to-file.toml renamed to tests/RobotFramework/tests/tedge_flows/flows/input-ext/append-to-file.toml

@@ -97,7 +97,6 @@ Running tedge-flows
     Execute Command    tedge mqtt sub test/count/e --duration 2s | grep '{}'
 
 Consuming messages from a process stdout
-    # Assuming the flow journalctl-follow.toml has been properly installed
     Install Journalctl Flow    journalctl-follow.toml
     ${test_start}    Get Unix Timestamp
     Restart Service    tedge-agent
@@ -106,7 +105,6 @@ Consuming messages from a process stdout
     [Teardown]    Uninstall Journalctl Flow    journalctl-follow.toml
 
 Consuming messages from a process stdout, periodically
-    # Assuming the flow journalctl-cursor.toml has been properly installed
     Install Journalctl Flow    journalctl-cursor.toml
     ${test_start}    Get Unix Timestamp
     Restart Service    tedge-agent
@@ -115,32 +113,35 @@ Consuming messages from a process stdout, periodically
     [Teardown]    Uninstall Journalctl Flow    journalctl-cursor.toml
 
 Consuming messages from the tail of file
-    # Assuming the flow tail-named-pipe.toml has been properly installed
+    Install Flow    tail-named-pipe.toml
     ${start}    Get Unix Timestamp
     Execute Command    echo hello>/tmp/events
     Should Have MQTT Messages    topic=log/events    message_contains=hello    minimum=1    date_from=${start}
+    [Teardown]    Uninstall Flow    tail-named-pipe.toml
 
 Consuming messages from a file, periodically
-    # Assuming the flow read-file-periodically.toml has been properly installed
+    Install Flow    read-file-periodically.toml
     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
     Execute Command    tedge mqtt sub test/file/input --duration 1s | grep world
     Execute Command    rm /tmp/file.input
     Execute Command
-    ...    tedge mqtt sub test/file/input --duration 1s | grep 'Error in /etc/tedge/flows/read-file-periodically.toml'
+    ...    tedge mqtt sub test/file/input --duration 1s | grep 'Fail to poll /tmp/file.input'
     Execute Command    echo 'hello world' >/tmp/file.input
     Execute Command    tedge mqtt sub test/file/input --duration 1s | grep 'hello world'
+    [Teardown]    Uninstall Flow    read-file-periodically.toml
 
 Appending messages to a file
-    # Assuming the flow append-to-file.toml has been properly installed
+    Install Flow    append-to-file.toml
     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
-    # Assuming the flow publish-js-errors.toml has been properly installed
+    Install Flow    publish-js-errors.toml
     ${start}    Get Unix Timestamp
     Execute Command    tedge mqtt pub collectd/foo 12345:6789
     Should Have MQTT Messages
@@ -167,6 +168,7 @@ Publishing transformation errors
     ${message}    JSONLibrary.Convert String To Json    ${messages[0]}
     Should Be Equal As Integers    ${message["time"]}    12345
     Should Be Equal As Integers    ${message["b"]["c"]}    6789
+    [Teardown]    Uninstall Flow    publish-js-errors.toml
 
 
 *** Keywords ***
@@ -188,6 +190,14 @@ Uninstall Journalctl Flow
     [Arguments]    ${definition_file}
     Execute Command    cmd=rm -f /etc/tedge/flows/${definition_file}
 
+Install Flow
+    [Arguments]    ${definition_file}
+    ThinEdgeIO.Transfer To Device    ${CURDIR}/input-ext/${definition_file}    /etc/tedge/flows/
+
+Uninstall Flow
+    [Arguments]    ${definition_file}
+    Execute Command    cmd=rm -f /etc/tedge/flows/${definition_file}
+
 Configure flows
     # Required by tail-named-pipe.toml
     Execute Command    mkfifo /tmp/events