wip(inventory-agent): local presence advertisement and discovery using mdns almost working

harmony_inventory_agent/Cargo.toml

@@ -10,3 +10,6 @@ serde.workspace = true
 serde_json.workspace = true
 log.workspace = true
 env_logger.workspace = true
+tokio.workspace = true
+thiserror.workspace = true
+mdns-sd = "0.14.1"

harmony_inventory_agent/src/lib.rs

@@ -0,0 +1,2 @@
+pub mod local_presence;
+mod hwinfo;

harmony_inventory_agent/src/local_presence/advertise.rs

@@ -0,0 +1,73 @@
+use log::{error, info, warn};
+use mdns_sd::{ServiceDaemon, ServiceInfo};
+use std::collections::HashMap;
+
+use crate::{hwinfo::PhysicalHost, local_presence::{PresenceError, SERVICE_NAME, VERSION}};
+
+/// Advertises the agent's presence on the local network.
+///
+/// This function is synchronous and non-blocking. It spawns a background Tokio task
+/// to handle the mDNS advertisement for the lifetime of the application.
+pub fn advertise(service_port: u16) -> Result<(), PresenceError> {
+    let host_id = match PhysicalHost::gather() {
+        Ok(host) => Some(host.host_uuid),
+        Err(e) => {
+            error!("Could not build physical host, harmony presence id will be unavailable : {e}");
+            None
+        }
+    };
+
+    let instance_name = format!("inventory-agent-{}", host_id.clone().unwrap_or("unknown".to_string()));
+
+    let spawned_msg = format!("Spawned local presence advertisement task for '{instance_name}'.");
+
+    tokio::spawn(async move {
+        info!(
+            "Local presence task started. Advertising as '{}'.",
+            instance_name
+        );
+
+        // The ServiceDaemon must live for the entire duration of the advertisement.
+        // If it's dropped, the advertisement stops.
+        let mdns = match ServiceDaemon::new() {
+            Ok(daemon) => daemon,
+            Err(e) => {
+                warn!("Failed to create mDNS daemon: {}. Task shutting down.", e);
+                return;
+            }
+        };
+
+        let mut props = HashMap::new();
+        if let Some(host_id) = host_id {
+            props.insert("id".to_string(), host_id);
+        }
+        props.insert("version".to_string(), VERSION.to_string());
+
+        let service_info = ServiceInfo::new(
+            SERVICE_NAME,
+            &instance_name,
+            &format!("{}.local.", instance_name),
+            (), // Let the daemon determine the host IPs
+            service_port,
+            Some(props),
+        )
+        .expect("ServiceInfo creation should not fail with valid inputs");
+
+        // The registration handle must also be kept alive.
+        let _registration_handle = match mdns.register(service_info) {
+            Ok(handle) => {
+                info!("Service successfully registered on the local network.");
+                handle
+            }
+            Err(e) => {
+                warn!("Failed to register service: {}. Task shutting down.", e);
+                return;
+            }
+        };
+
+    });
+
+    info!("{spawned_msg}");
+
+    Ok(())
+}

harmony_inventory_agent/src/local_presence/discover.rs

@@ -0,0 +1,34 @@
+use mdns_sd::{ServiceDaemon, ServiceEvent};
+
+use crate::local_presence::SERVICE_NAME;
+
+pub type DiscoveryEvent = ServiceEvent;
+
+pub fn discover_agents(timeout: Option<u64>, on_event: fn(&DiscoveryEvent)) {
+    // Create a new mDNS daemon.
+    let mdns = ServiceDaemon::new().expect("Failed to create mDNS daemon");
+
+    // Start browsing for the service type.
+    // The receiver will be a stream of events.
+    let receiver = mdns.browse(SERVICE_NAME).expect("Failed to browse");
+
+    std::thread::spawn(move || {
+        while let Ok(event) = receiver.recv() {
+            on_event(&event);
+            match event {
+                ServiceEvent::ServiceData(resolved) => {
+                    println!("Resolved a new service: {}", resolved.fullname);
+                }
+                other_event => {
+                    println!("Received other event: {:?}", &other_event);
+                }
+            }
+        }
+    });
+
+    if let Some(timeout) = timeout {
+    // Gracefully shutdown the daemon.
+    std::thread::sleep(std::time::Duration::from_secs(timeout));
+    mdns.shutdown().unwrap();
+    }
+}

harmony_inventory_agent/src/local_presence/mod.rs

@@ -0,0 +1,16 @@
+mod discover;
+pub use discover::*;
+mod advertise;
+pub use advertise::*;
+
+pub const SERVICE_NAME: &str = "_harmony._tcp.local.";
+const VERSION: &str = env!("CARGO_PKG_VERSION");
+
+// A specific error type for our module enhances clarity and usability.
+#[derive(thiserror::Error, Debug)]
+pub enum PresenceError {
+    #[error("Failed to create mDNS daemon")]
+    DaemonCreationFailed(#[from] mdns_sd::Error),
+    #[error("The shutdown signal has already been sent")]
+    ShutdownFailed,
+}

harmony_inventory_agent/src/main.rs

@@ -1,9 +1,15 @@
 // src/main.rs
 use actix_web::{App, HttpServer, Responder, get};
-use hwinfo::PhysicalHost;
+use log::error;
 use std::env;
 
+use crate::hwinfo::PhysicalHost;
+
 mod hwinfo;
+mod local_presence;
+
+
+
 
 #[get("/inventory")]
 async fn inventory() -> impl Responder {
@@ -26,10 +32,15 @@ async fn main() -> std::io::Result<()> {
     env_logger::init();
 
     let port = env::var("HARMONY_INVENTORY_AGENT_PORT").unwrap_or_else(|_| "8080".to_string());
+    let port = port.parse::<u16>().expect(&format!("Invalid port number, cannot parse to u16 {port}"));
     let bind_addr = format!("0.0.0.0:{}", port);
 
     log::info!("Starting inventory agent on {}", bind_addr);
 
+    if let Err(e) = local_presence::advertise(port) {
+        error!("Could not start advertise local presence : {e}");
+    }
+
     HttpServer::new(|| App::new().service(inventory))
         .bind(&bind_addr)?
         .run()