use flume::RecvTimeoutError;
use mdns_sd::{ServiceDaemon, ServiceInfo};
use std::net::IpAddr;
use std::sync::{Arc, Mutex};
use std::time::Duration;
#[derive(Clone, Debug)]
pub struct PeerDiscovery {
pub node_id: String,
pub hostname: String,
pub ip: IpAddr,
pub port: u16,
}
pub struct MdnsDiscovery {
node_id: String,
port: u16,
service_type: String,
daemon: Arc<Mutex<Option<ServiceDaemon>>>,
}
impl MdnsDiscovery {
const SERVICE_TYPE: &'static str = "_pot-o-validator._tcp.local.";
pub fn new(node_id: &str, port: u16) -> Result<Self, Box<dyn std::error::Error>> {
let daemon =
ServiceDaemon::new().map_err(|e| format!("Failed to create mDNS daemon: {}", e))?;
Ok(Self {
node_id: node_id.to_string(),
port,
service_type: Self::SERVICE_TYPE.to_string(),
daemon: Arc::new(Mutex::new(Some(daemon))),
})
}
pub fn node_id(&self) -> &str {
&self.node_id
}
pub fn port(&self) -> u16 {
self.port
}
pub fn service_type(&self) -> &str {
&self.service_type
}
pub fn register_service(&self, hostname: &str) -> Result<(), Box<dyn std::error::Error>> {
let daemon_lock = self
.daemon
.lock()
.map_err(|e| format!("Failed to lock daemon: {}", e))?;
let daemon = daemon_lock.as_ref().ok_or("mDNS daemon not initialized")?;
let service_info = ServiceInfo::new(
Self::SERVICE_TYPE,
&format!("{}._pot-o-validator._tcp.local.", self.node_id),
&self.node_id,
hostname,
self.port,
None,
)
.map_err(|e| format!("Failed to create ServiceInfo: {}", e))?;
daemon
.register(service_info)
.map_err(|e| format!("Failed to register service: {}", e))?;
Ok(())
}
pub fn discover_peers(
&self,
timeout_secs: u64,
) -> Result<Vec<PeerDiscovery>, Box<dyn std::error::Error>> {
let daemon_lock = self
.daemon
.lock()
.map_err(|e| format!("Failed to lock daemon: {}", e))?;
let daemon = daemon_lock.as_ref().ok_or("mDNS daemon not initialized")?;
let receiver = daemon
.browse(Self::SERVICE_TYPE)
.map_err(|e| format!("Failed to browse services: {}", e))?;
let mut peers = Vec::new();
let deadline = std::time::Instant::now() + Duration::from_secs(timeout_secs);
loop {
let remaining = deadline.saturating_duration_since(std::time::Instant::now());
if remaining.is_zero() {
break;
}
match receiver.recv_timeout(remaining) {
Ok(mdns_sd::ServiceEvent::ServiceResolved(info)) => {
if let Some(peer) = self.extract_peer_from_service(&info) {
if peer.node_id != self.node_id {
peers.push(peer);
}
}
}
Ok(mdns_sd::ServiceEvent::ServiceRemoved(_, _)) => {
}
Ok(mdns_sd::ServiceEvent::SearchStarted(_)) => {
}
Ok(mdns_sd::ServiceEvent::ServiceFound(_, _)) => {
}
Ok(mdns_sd::ServiceEvent::SearchStopped(_)) => {
break;
}
Err(RecvTimeoutError::Timeout) => {
break; }
Err(_) => {
break;
}
}
}
Ok(peers)
}
pub fn unregister_service(&self) -> Result<(), Box<dyn std::error::Error>> {
let daemon_lock = self
.daemon
.lock()
.map_err(|e| format!("Failed to lock daemon: {}", e))?;
let daemon = daemon_lock.as_ref().ok_or("mDNS daemon not initialized")?;
let service_name = format!("{}._pot-o-validator._tcp.local.", self.node_id);
daemon
.unregister(&service_name)
.map_err(|e| format!("Failed to unregister service: {}", e))?;
Ok(())
}
fn extract_peer_from_service(&self, info: &ServiceInfo) -> Option<PeerDiscovery> {
let fullname = info.get_fullname();
let node_id = fullname
.strip_suffix("._pot-o-validator._tcp.local.")?
.to_string();
let hostname = info.get_hostname().to_string();
let ip = info
.get_addresses()
.iter()
.next()
.map(|ipv4| IpAddr::V4(*ipv4))?;
let port = info.get_port();
Some(PeerDiscovery {
node_id,
hostname,
ip,
port,
})
}
}
#[allow(dead_code)]
fn _assert_send_sync()
where
MdnsDiscovery: Send + Sync,
{
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_service_type_constant() {
assert_eq!(MdnsDiscovery::SERVICE_TYPE, "_pot-o-validator._tcp.local.");
}
#[test]
fn test_peer_discovery_creation() {
let peer = PeerDiscovery {
node_id: "test-node".to_string(),
hostname: "test-node.local".to_string(),
ip: "127.0.0.1".parse().unwrap(),
port: 5555,
};
assert_eq!(peer.node_id, "test-node");
assert_eq!(peer.port, 5555);
}
}