refactor(networkmanager): merge devices and watchers fields in ClientInner

src/clients/networkmanager/mod.rs

@@ -45,28 +45,25 @@ impl Client {
 struct ClientInner {
     controller_sender: broadcast::Sender<ClientToModuleEvent>,
     sender: broadcast::Sender<ModuleToClientEvent>,
-    devices: RwLock<HashSet<ObjectPath<'static>>>,
+    device_watchers: RwLock<HashMap<ObjectPath<'static>, DeviceWatcher>>,
-    watchers: RwLock<HashMap<ObjectPath<'static>, Device>>,
     // TODO: Maybe find some way to late-init a dbus connection here
     //     so we can just clone it when we need it instead of awaiting it every time
 }
 
-#[derive(Debug)]
+#[derive(Clone, Debug)]
-struct Device {
+struct DeviceWatcher {
-    state_watcher: JoinHandle<Result<()>>,
+    state_watcher: Arc<JoinHandle<Result<()>>>,
 }
 
 impl ClientInner {
     fn new() -> ClientInner {
         let (controller_sender, _) = broadcast::channel(64);
         let (sender, _) = broadcast::channel(8);
-        let devices = RwLock::new(HashSet::new());
+        let device_watchers = RwLock::new(HashMap::new());
-        let watchers = RwLock::new(HashMap::new());
         ClientInner {
             controller_sender,
             sender,
-            devices,
+            device_watchers,
-            watchers,
         }
     }
 
@@ -98,35 +95,34 @@ impl ClientInner {
         let mut devices_changes = root.receive_all_devices_changed().await;
         while let Some(devices_change) = devices_changes.next().await {
             // The new list of devices from dbus, not to be confused with the added devices below
-            let new_devices = devices_change
+            let new_device_paths = devices_change
                 .get()
                 .await?
                 .iter()
                 .map(ObjectPath::to_owned)
                 .collect::<HashSet<_>>();
 
-            // TODO: Use `self.watchers` instead of `self.devices`, which requires creating all property watchers straightaway
+            let mut watchers = self.device_watchers.write().await;
+            let device_paths = watchers.keys().cloned().collect::<HashSet<_>>();
 
-            // Atomic read-then-write of `devices`
+            let added_device_paths = new_device_paths.difference(&device_paths);
-            let mut devices_locked = self.devices.write().await;
+            for added_device_path in added_device_paths {
-            let devices_snapshot = devices_locked.clone();
+                debug_assert!(!watchers.contains_key(added_device_path));
-            (*devices_locked).clone_from(&new_devices);
-            drop(devices_locked);
 
-            let added_devices = new_devices.difference(&devices_snapshot);
+                let watcher = self.watch_device(added_device_path.clone());
-            for added_device in added_devices {
+                watchers.insert(added_device_path.clone(), watcher);
-                spawn(self.watch_device(added_device.to_owned()));
             }
 
             // TODO: Inform module of removed devices
-            let removed_devices = devices_snapshot.difference(&new_devices);
+            let removed_device_paths = device_paths.difference(&new_device_paths);
-            for removed_device in removed_devices {
+            for removed_device_path in removed_device_paths {
-                let mut watchers = self.watchers.write().await;
+                let watcher = watchers
-                let device = watchers.get(removed_device).unwrap();
+                    .get(removed_device_path)
-                device.state_watcher.abort();
+                    .expect("Device to be removed should be present in watchers");
-                watchers.remove(removed_device);
+                watcher.state_watcher.abort();
+                watchers.remove(removed_device_path);
 
-                debug!("D-bus device state watcher for {} stopped", removed_device);
+                debug!("D-bus device watchers for {} stopped", removed_device_path);
             }
         }
 
@@ -144,10 +140,7 @@ impl ClientInner {
                 ModuleToClientEvent::NewController => {
                     debug!("Client received NewController event");
 
-                    // We create a local clone here to avoid holding the lock for too long
+                    for device_path in self.device_watchers.read().await.keys() {
-                    let devices_snapshot = self.devices.read().await.clone();
-
-                    for device_path in devices_snapshot {
                         let device = DeviceDbusProxy::new(&dbus_connection, device_path).await?;
 
                         let interface = device.interface().await?.to_string();
@@ -167,24 +160,18 @@ impl ClientInner {
         Ok(())
     }
 
-    async fn watch_device(&'static self, path: ObjectPath<'static>) -> Result<()> {
+    fn watch_device(&'static self, path: ObjectPath<'static>) -> DeviceWatcher {
-        debug_assert!(!self.watchers.read().await.contains_key(&path));
+        let state_watcher = Arc::new(spawn(self.watch_device_state(path)));
 
-        let state_watcher = spawn(self.watch_device_state(path.clone()));
+        DeviceWatcher { state_watcher }
-        self.watchers
-            .write()
-            .await
-            .insert(path, Device { state_watcher });
-
-        Ok(())
     }
 
     async fn watch_device_state(&'static self, path: ObjectPath<'_>) -> Result<()> {
+        debug!("D-Bus device state watcher for {} starting", path);
+
         let dbus_connection = Connection::system().await?;
         let device = DeviceDbusProxy::new(&dbus_connection, path.clone()).await?;
 
-        debug!("D-Bus device state watcher for {} starting", path);
-
         let interface = device.interface().await?;
         let r#type = device.device_type().await?;
 
@@ -208,8 +195,6 @@ impl ClientInner {
                 })?;
         }
 
-        debug!("D-Bus device state watcher for {} ended", path);
-
         Ok(())
     }
 }