Dynamic discovery of new operations (#1140)

* Closes #612 discover operations dynamically

Signed-off-by: Pradeep Kumar K J <pradeepkumar.kj@softwareag.com>

* select! on async ops

Signed-off-by: Pradeep Kumar K J <pradeepkumar.kj@softwareag.com>

* update operations document

Signed-off-by: Pradeep Kumar K J <pradeepkumar.kj@softwareag.com>

* move inotify/mqtt message processing to separate fn

* move mqtt process code to a separate fn

2 files changed, 34 insertions, 10 deletions

diff --git a/crates/core/c8y_smartrest/Cargo.toml b/crates/core/c8y_smartrest/Cargo.toml
index 78d7a9bf..df288f79 100644
--- a/crates/core/c8y_smartrest/Cargo.toml
+++ b/crates/core/c8y_smartrest/Cargo.toml
@@ -18,6 +18,7 @@ thiserror = "1.0"
 time = { version = "0.3", features = ["formatting", "macros", "parsing", "serde"] }
 tokio = { version = "1.8", features = ["rt", "sync", "time"] }
 toml = "0.5"
+tracing = { version = "0.1", features = ["attributes", "log"] }
 
 [dev-dependencies]
 anyhow = "1.0"
diff --git a/crates/core/c8y_smartrest/src/operations.rs b/crates/core/c8y_smartrest/src/operations.rs
index c7bc7c59..5304b8b2 100644
--- a/crates/core/c8y_smartrest/src/operations.rs
+++ b/crates/core/c8y_smartrest/src/operations.rs
@@ -4,9 +4,8 @@ use std::{
     path::{Path, PathBuf},
 };
 
-use serde::Deserialize;
-
 use crate::error::OperationsError;
+use serde::Deserialize;
 
 /// Operations are derived by reading files subdirectories per cloud /etc/tedge/operations directory
 /// Each operation is a file name in one of the subdirectories
@@ -59,14 +58,22 @@ impl Default for Operations {
 }
 
 impl Operations {
-    pub fn add(&mut self, operation: Operation) {
-        if let Some(detail) = operation.exec() {
-            if let Some(on_message) = &detail.on_message {
-                self.operations_by_trigger
-                    .insert(on_message.clone(), self.operations.len());
+    pub fn add_operation(&mut self, operation: Operation) {
+        if self.operations.iter().any(|o| o.name.eq(&operation.name)) {
+            return;
+        } else {
+            if let Some(detail) = operation.exec() {
+                if let Some(on_message) = &detail.on_message {
+                    self.operations_by_trigger
+                        .insert(on_message.clone(), self.operations.len());
+                }
             }
+            self.operations.push(operation);
         }
-        self.operations.push(operation);
+    }
+
+    pub fn remove_operation(&mut self, op_name: &str) {
+        self.operations.retain(|x| x.name.ne(&op_name));
     }
 
     pub fn try_new(dir: impl AsRef<Path>, cloud_name: &str) -> Result<Self, OperationsError> {
@@ -121,12 +128,28 @@ fn get_operations(dir: impl AsRef<Path>, cloud_name: &str) -> Result<Operations,
             .and_then(|filename| filename.to_str())
             .ok_or_else(|| OperationsError::InvalidOperationName(path.to_owned()))?
             .to_owned();
-
-        operations.add(details);
+        operations.add_operation(details);
     }
     Ok(operations)
 }
 
+pub fn get_operation(path: PathBuf) -> Result<Operation, OperationsError> {
+    let mut details = match fs::read(&path) {
+        Ok(bytes) => toml::from_slice::<Operation>(bytes.as_slice())
+            .map_err(|e| OperationsError::TomlError(path.to_path_buf(), e))?,
+
+        Err(err) => return Err(OperationsError::FromIo(err)),
+    };
+
+    details.name = path
+        .file_name()
+        .and_then(|filename| filename.to_str())
+        .ok_or_else(|| OperationsError::InvalidOperationName(path.to_owned()))?
+        .to_owned();
+
+    Ok(details)
+}
+
 #[cfg(test)]
 mod tests {
     use std::io::Write;