refactor: simplify 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.9"
+version = "0.41.10"
 
 [dependencies]
 actix-server = { version = "2.6", default-features = false }
@@ -25,12 +25,10 @@ freedesktop_entry_parser = { version = "1.3", default-features = false }
 futures = { version = "0.3", default-features = false }
 futures-util = { version = "0.3", default-features = false }
 glob = { version = "0.3", default-features = false }
-lazy_static = { version = "1.5", default-features = false }
+inotify = { version = "0.11", default-features = false, features = ["stream"] }
 log = { version = "0.4", default-features = false }
 log-panics = { version = "2", default-features = false }
 modemmanager = { git = "https://github.com/omnect/modemmanager.git", tag = "0.3.4", default-features = false, optional = true }
-notify = { version = "8.0", default-features = false }
-notify-debouncer-full = { version = "0.5", default-features = false }
 regex-lite = { version = "0.1", default-features = true }
 reqwest = { version = "0.12", default-features = false, features = [
   "default-tls",

Cargo.toml

@@ -1,19 +1,19 @@
 use crate::{
     common::{from_json_file, to_json_file},
-    twin::{Feature, feature::*},
+    twin::feature::*,
 };
 use anyhow::{Context, Result, bail, ensure};
 use azure_iot_sdk::client::IotMessage;
+use inotify::WatchMask;
 use log::{info, warn};
-use notify_debouncer_full::{Debouncer, NoCache, notify::*};
 use serde::{Deserialize, Serialize};
 use serde_json::json;
-use std::{collections::HashMap, env, path::Path};
+use std::{collections::HashMap, env, ffi::c_int, path::PathBuf};
 use tokio::sync::mpsc::Sender;
 
 macro_rules! consent_path {
     () => {
-        Path::new(&env::var("CONSENT_DIR_PATH").unwrap_or("/etc/omnect/consent".to_string()))
+        PathBuf::from(&env::var("CONSENT_DIR_PATH").unwrap_or("/etc/omnect/consent".to_string()))
     };
 }
 
@@ -53,9 +53,8 @@ pub struct ConsentConfig {
     reset_consent_on_fail: bool,
 }
 
-#[derive(Default)]
 pub struct DeviceUpdateConsent {
-    file_observer: Option<Debouncer<INotifyWatcher, NoCache>>,
+    file_observer: HashMap<c_int, PathBuf>,
     tx_reported_properties: Option<Sender<serde_json::Value>>,
 }
 
@@ -87,28 +86,23 @@ 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) => {
-                self.report_consent(from_json_file(&file.path)?).await?;
+            Command::WatchPath(cmd) => {
+                self.report_consent(from_json_file(
+                    self.file_observer
+                        .get(&cmd.id)
+                        .context("cannot find file for watch id")?,
+                )?)
+                .await?;
             }
             Command::DesiredGeneralConsent(cmd) => {
                 self.update_general_consent(cmd).await?;
             }
             Command::UserConsent(cmd) => {
                 self.user_consent(cmd)?;
             }
-            _ => bail!("unexpected command"),
+            _ => bail!("unexpected command: {cmd:?}"),
         };
 
         Ok(None)
@@ -119,6 +113,24 @@ impl DeviceUpdateConsent {
     const USER_CONSENT_VERSION: u8 = 1;
     const ID: &'static str = "device_update_consent";
 
+    pub async fn new() -> Result<Self> {
+        let mut file_observer = HashMap::new();
+
+        for path in [request_consent_path!(), history_consent_path!()] {
+            file_observer.insert(
+                add_fs_watch::<Self>(&path, WatchMask::MODIFY)
+                    .await?
+                    .get_watch_descriptor_id(),
+                path,
+            );
+        }
+
+        Ok(DeviceUpdateConsent {
+            tx_reported_properties: None,
+            file_observer,
+        })
+    }
+
     fn user_consent(&self, cmd: &UserConsentCommand) -> CommandResult {
         info!("user consent requested: {cmd:?}");
 
@@ -192,27 +204,29 @@ mod tests {
     async fn consent_files_changed_test() {
         let (tx_reported_properties, mut rx_reported_properties) = tokio::sync::mpsc::channel(100);
         let mut consent = DeviceUpdateConsent {
-            file_observer: None,
+            file_observer: HashMap::from([(
+                1,
+                PathBuf::from("testfiles/positive/test_component/user_consent.json"),
+            )]),
             tx_reported_properties: Some(tx_reported_properties),
         };
 
         assert!(
             consent
-                .command(&Command::FileModified(PathCommand {
+                .command(&Command::WatchPath(WatchPathCommand {
                     feature_id: TypeId::of::<DeviceUpdateConsent>(),
-                    path: Path::new("my-path").to_path_buf(),
+                    id: 0,
                 }))
                 .await
                 .unwrap_err()
-                .chain()
-                .any(|e| e.to_string().starts_with("failed to open for read: "))
+                .to_string()
+                .eq("cannot find file for watch id")
         );
 
         consent
-            .command(&Command::FileModified(PathCommand {
+            .command(&Command::WatchPath(WatchPathCommand {
                 feature_id: TypeId::of::<DeviceUpdateConsent>(),
-                path: Path::new("testfiles/positive/test_component/user_consent.json")
-                    .to_path_buf(),
+                id: 1,
             }))
             .await
             .unwrap();
@@ -227,7 +241,7 @@ mod tests {
     async fn desired_consent_test() {
         let (tx_reported_properties, mut rx_reported_properties) = tokio::sync::mpsc::channel(100);
         let mut consent = DeviceUpdateConsent {
-            file_observer: None,
+            file_observer: HashMap::new(),
             tx_reported_properties: Some(tx_reported_properties),
         };
 
@@ -318,7 +332,7 @@ mod tests {
     async fn user_consent_test() {
         let (tx_reported_properties, _rx_reported_properties) = tokio::sync::mpsc::channel(100);
         let mut consent = DeviceUpdateConsent {
-            file_observer: None,
+            file_observer: HashMap::new(),
             tx_reported_properties: Some(tx_reported_properties),
         };