refactor: CommandRequest handling

Signed-off-by: Jan Zachmann <50990105+JanZachmann@users.noreply.github.com>

Author: JanZachmann

Cargo.lock

@@ -8,7 +8,7 @@ license = "MIT OR Apache-2.0"
 name = "omnect-device-service"
 readme = "README.md"
 repository = "https://github.com/omnect/omnect-device-service.git"
-version = "0.41.5"
+version = "0.41.6"
 
 [dependencies]
 actix-server = { version = "2.6", default-features = false }
@@ -64,6 +64,7 @@ tokio-stream = { version = "0.1", default-features = false, features = [
   "time",
 ] }
 toml = { version = "0.8", default-features = false, features = ["parse"] }
+typeid = { version = "1.0", default-features = false }
 uuid = { version = "1.17", default-features = false }
 x509-parser = { version = "0.17", default-features = false }
 zbus = { version = "5.7", default-features = false, features = ["tokio"] }

Cargo.toml

@@ -5,7 +5,7 @@ use crate::{
 use anyhow::{Context, Result, bail, ensure};
 use azure_iot_sdk::client::IotMessage;
 use log::{info, warn};
-use notify_debouncer_full::{Debouncer, NoCache, notify::*};
+use notify_debouncer_full::notify::*;
 use serde::{Deserialize, Serialize};
 use serde_json::json;
 use std::{collections::HashMap, env, path::Path};
@@ -53,9 +53,7 @@ pub struct ConsentConfig {
     reset_consent_on_fail: bool,
 }
 
-#[derive(Default)]
 pub struct DeviceUpdateConsent {
-    file_observer: Option<Debouncer<INotifyWatcher, NoCache>>,
     tx_reported_properties: Option<Sender<serde_json::Value>>,
 }
 
@@ -87,19 +85,9 @@ impl Feature for DeviceUpdateConsent {
             .await
     }
 
-    fn command_request_stream(&mut self) -> CommandRequestStreamResult {
-        let (file_observer, stream) = file_modified_stream::<DeviceUpdateConsent>(vec![
-            request_consent_path!().as_path(),
-            history_consent_path!().as_path(),
-        ])
-        .context("command_request_stream: cannot create file_modified_stream")?;
-        self.file_observer = Some(file_observer);
-        Ok(Some(stream))
-    }
-
     async fn command(&mut self, cmd: &Command) -> CommandResult {
         match cmd {
-            Command::FileModified(file) => {
+            Command::WatchPath(file) => {
                 self.report_consent(from_json_file(&file.path)?).await?;
             }
             Command::DesiredGeneralConsent(cmd) => {
@@ -119,6 +107,26 @@ impl DeviceUpdateConsent {
     const USER_CONSENT_VERSION: u8 = 1;
     const ID: &'static str = "device_update_consent";
 
+    pub fn new() -> Result<Self> {
+        for path in [request_consent_path!(), history_consent_path!()] {
+            watch_path(
+                Watch {
+                    command: PathCommand {
+                        feature_id: typeid::ConstTypeId::of::<Self>(),
+                        path,
+                    },
+                    event_kinds: vec![EventKind::Modify(event::ModifyKind::Data(
+                        event::DataChange::Content,
+                    ))],
+                },
+                RecursiveMode::Recursive,
+            )?;
+        }
+        Ok(DeviceUpdateConsent {
+            tx_reported_properties: None,
+        })
+    }
+
     fn user_consent(&self, cmd: &UserConsentCommand) -> CommandResult {
         info!("user consent requested: {cmd:?}");
 

src/twin/consent.rs

@@ -8,7 +8,7 @@ use crate::{
 use anyhow::{Context, Result, bail};
 use azure_iot_sdk::client::IotMessage;
 use log::{debug, info, warn};
-use notify_debouncer_full::{Debouncer, NoCache, notify::*};
+use notify_debouncer_full::notify::*;
 use serde::{Deserialize, Serialize};
 use serde_json::{from_reader, json};
 use serde_repr::*;
@@ -17,7 +17,7 @@ use std::{
     env,
     fs::{File, read_dir},
     io::BufReader,
-    path::Path,
+    path::PathBuf,
 };
 use tokio::sync::mpsc::Sender;
 
@@ -37,8 +37,10 @@ macro_rules! config_path {
 
 macro_rules! custom_config_dir_path {
     () => {
-        env::var("FACTORY_RESET_CUSTOM_CONFIG_DIR_PATH")
-            .unwrap_or("/etc/omnect/factory-reset.d".to_string())
+        PathBuf::from(
+            env::var("FACTORY_RESET_CUSTOM_CONFIG_DIR_PATH")
+                .unwrap_or("/etc/omnect/factory-reset.d".to_string()),
+        )
     };
 }
 
@@ -92,7 +94,6 @@ struct FactoryResetReport {
 pub struct FactoryReset {
     tx_reported_properties: Option<Sender<serde_json::Value>>,
     report: FactoryResetReport,
-    dir_observer: Option<Debouncer<INotifyWatcher, NoCache>>,
 }
 
 impl Feature for FactoryReset {
@@ -136,17 +137,9 @@ impl Feature for FactoryReset {
         Ok(())
     }
 
-    fn command_request_stream(&mut self) -> CommandRequestStreamResult {
-        let (dir_observer, stream) =
-            dir_modified_stream::<FactoryReset>(vec![&Path::new(&custom_config_dir_path!())])
-                .context("command_request_stream: cannot create dir_modified_stream")?;
-        self.dir_observer = Some(dir_observer);
-        Ok(Some(stream))
-    }
-
     async fn command(&mut self, cmd: &Command) -> CommandResult {
         match cmd {
-            Command::DirModified(_) => {
+            Command::WatchPath(_) => {
                 let keys = FactoryReset::factory_reset_keys()?;
 
                 if keys != self.report.keys {
@@ -193,10 +186,23 @@ impl FactoryReset {
             result: FactoryReset::factory_reset_result()?,
         };
 
+        watch_path(
+            Watch {
+                command: PathCommand {
+                    feature_id: typeid::ConstTypeId::of::<Self>(),
+                    path: custom_config_dir_path!(),
+                },
+                event_kinds: vec![
+                    EventKind::Create(event::CreateKind::File),
+                    EventKind::Remove(event::RemoveKind::File),
+                ],
+            },
+            RecursiveMode::Recursive,
+        )?;
+
         Ok(FactoryReset {
             tx_reported_properties: None,
             report,
-            dir_observer: None,
         })
     }