fix(networkmanager): support late module initialisation

For example when a second monitor is connected while Ironbar is already running.

src/clients/networkmanager/mod.rs

@@ -3,13 +3,13 @@ use color_eyre::eyre::Ok;
 use futures_lite::StreamExt;
 use std::collections::HashSet;
 use std::sync::Arc;
-use tokio::sync::broadcast;
+use tokio::sync::{RwLock, broadcast};
 use zbus::Connection;
 use zbus::zvariant::{ObjectPath, Str};
 
 use crate::clients::ClientResult;
 use crate::clients::networkmanager::dbus::{DbusProxy, DeviceDbusProxy};
-use crate::clients::networkmanager::event::Event;
+use crate::clients::networkmanager::event::{ClientToModuleEvent, ModuleToClientEvent};
 use crate::{register_fallible_client, spawn};
 
 pub mod dbus;
@@ -17,47 +17,104 @@ pub mod event;
 
 #[derive(Debug)]
 pub struct Client {
-    tx: broadcast::Sender<Event>,
+    controller_sender: broadcast::Sender<ClientToModuleEvent>,
+    sender: broadcast::Sender<ModuleToClientEvent>,
 }
 
 impl Client {
     fn new() -> Result<Client> {
-        let (tx, _) = broadcast::channel(64);
-        Ok(Client { tx })
+        let (controller_sender, _) = broadcast::channel(64);
+        let (sender, _) = broadcast::channel(8);
+        Ok(Client {
+            controller_sender,
+            sender,
+        })
     }
 
     fn run(&self) -> Result<()> {
-        let tx = self.tx.clone();
-        spawn(async move {
-            let dbus_connection = Connection::system().await?;
-            let root = DbusProxy::new(&dbus_connection).await?;
+        let devices: &'static _ = Box::leak(Box::new(RwLock::new(HashSet::<ObjectPath>::new())));
 
-            let mut devices = HashSet::new();
+        {
+            let controller_sender = self.controller_sender.clone();
+            spawn(async move {
+                let dbus_connection = Connection::system().await?;
+                let root = DbusProxy::new(&dbus_connection).await?;
 
-            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 = HashSet::from_iter(devices_change.get().await?);
+                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
+                        .get()
+                        .await?
+                        .iter()
+                        .map(ObjectPath::to_owned)
+                        .collect::<HashSet<_>>();
 
-                let added_devices = new_devices.difference(&devices);
-                for added_device in added_devices {
-                    spawn(watch_device(added_device.to_owned(), tx.clone()));
+                    // We create a local clone here to avoid holding the lock for too long
+                    let devices_snapshot = devices.read().await.clone();
+
+                    let added_devices = new_devices.difference(&devices_snapshot);
+                    for added_device in added_devices {
+                        spawn(watch_device(
+                            added_device.to_owned(),
+                            controller_sender.clone(),
+                        ));
+                    }
+
+                    let _removed_devices = devices_snapshot.difference(&new_devices);
+                    // TODO: Cook up some way to notify closures for removed devices to exit
+
+                    *devices.write().await = new_devices;
                 }
 
-                let _removed_devices = devices.difference(&new_devices);
-                // TODO: Cook up some way to notify closures for removed devices to exit
+                Ok(())
+            });
+        }
 
-                devices = new_devices;
-            }
+        {
+            let controller_sender = self.controller_sender.clone();
+            let mut receiver = self.sender.subscribe();
+            spawn(async move {
+                while let Result::Ok(event) = receiver.recv().await {
+                    match event {
+                        ModuleToClientEvent::NewController => {
+                            // We create a local clone here to avoid holding the lock for too long
+                            let devices_snapshot = devices.read().await.clone();
 
-            Ok(())
-        });
+                            for device_path in devices_snapshot {
+                                let dbus_connection = Connection::system().await?;
+                                let device =
+                                    DeviceDbusProxy::new(&dbus_connection, device_path).await?;
+
+                                // TODO: Create DeviceDbusProxy -> DeviceStateChanged function and use it in the watcher as well
+                                let interface = device.interface().await?.to_string();
+                                let r#type = device.device_type().await?;
+                                let state = device.state().await?;
+                                controller_sender.send(
+                                    ClientToModuleEvent::DeviceStateChanged {
+                                        interface,
+                                        r#type,
+                                        state,
+                                    },
+                                )?;
+                            }
+                        }
+                    }
+                }
+
+                Ok(())
+            });
+        }
 
         Ok(())
     }
 
-    pub fn subscribe(&self) -> broadcast::Receiver<Event> {
-        self.tx.subscribe()
+    pub fn subscribe(&self) -> broadcast::Receiver<ClientToModuleEvent> {
+        self.controller_sender.subscribe()
+    }
+
+    pub fn get_sender(&self) -> broadcast::Sender<ModuleToClientEvent> {
+        self.sender.clone()
     }
 }
 
@@ -67,19 +124,19 @@ pub fn create_client() -> ClientResult<Client> {
     Ok(client)
 }
 
-async fn watch_device(device_path: ObjectPath<'_>, tx: broadcast::Sender<Event>) -> Result<()> {
+async fn watch_device(
+    device_path: ObjectPath<'_>,
+    controller_sender: broadcast::Sender<ClientToModuleEvent>,
+) -> Result<()> {
     let dbus_connection = Connection::system().await?;
     let device = DeviceDbusProxy::new(&dbus_connection, device_path.to_owned()).await?;
 
     let interface = device.interface().await?;
-    tx.send(Event::DeviceAdded {
-        interface: interface.to_string(),
-    })?;
 
     spawn(watch_device_state(
         device_path.to_owned(),
         interface.to_owned(),
-        tx.clone(),
+        controller_sender.clone(),
     ));
 
     Ok(())
@@ -88,7 +145,7 @@ async fn watch_device(device_path: ObjectPath<'_>, tx: broadcast::Sender<Event>)
 async fn watch_device_state(
     device_path: ObjectPath<'_>,
     interface: Str<'_>,
-    tx: broadcast::Sender<Event>,
+    controller_sender: broadcast::Sender<ClientToModuleEvent>,
 ) -> Result<()> {
     let dbus_connection = Connection::system().await?;
     let device = DeviceDbusProxy::new(&dbus_connection, &device_path).await?;
@@ -96,7 +153,7 @@ async fn watch_device_state(
 
     // Send an event communicating the initial state
     let state = device.state().await?;
-    tx.send(Event::DeviceStateChanged {
+    controller_sender.send(ClientToModuleEvent::DeviceStateChanged {
         interface: interface.to_string(),
         r#type: r#type.clone(),
         state,
@@ -105,7 +162,7 @@ async fn watch_device_state(
     let mut state_changes = device.receive_state_changed().await;
     while let Some(state_change) = state_changes.next().await {
         let state = state_change.get().await?;
-        tx.send(Event::DeviceStateChanged {
+        controller_sender.send(ClientToModuleEvent::DeviceStateChanged {
             interface: interface.to_string(),
             r#type: r#type.clone(),
             state,