use std::net::SocketAddrV4; use anyhow::{Result, Context}; use igd::aio::*; use igd::PortMappingProtocol; use log::*; use tokio::{ select, sync::watch, time::{ Duration, self, }}; use crate::config::RuntimeConfigIgd; use crate::messages; pub struct IgdActor { expire: Duration, gateway: Gateway, last_ports: messages::PublicExposedPorts, private_ip: String, refresh: Duration, rx_ports: watch::Receiver, } impl IgdActor { pub async fn new(config: Option, rxp: &watch::Receiver) -> Result> { if config.is_none() { return Ok(None); } let config = config.unwrap(); let gw = search_gateway(Default::default()) .await .context("Failed to find IGD gateway")?; info!("IGD gateway: {}", gw); let ctx = Self { expire: config.expiration_time, gateway: gw, last_ports: messages::PublicExposedPorts::new(), private_ip: config.private_ip, refresh: config.refresh_time, rx_ports: rxp.clone(), }; return Ok(Some(ctx)); } pub async fn listen(&mut self) -> Result<()> { let mut interval = time::interval(self.refresh); loop { // 1. Wait for an event let new_ports = select! { Some(ports) = self.rx_ports.recv() => Some(ports), _ = interval.tick() => None, else => return Ok(()) // Sender dropped, terminate loop. }; // 2. Update last ports if needed if let Some(p) = new_ports { self.last_ports = p; } // 3. Flush IGD requests match self.do_igd().await { Ok(()) => debug!("Successfully updated IGD"), Err(e) => error!("An error occured while updating IGD. {}", e), } } } pub async fn do_igd(&self) -> Result<()> { let actions = [ (PortMappingProtocol::TCP, &self.last_ports.tcp_ports), (PortMappingProtocol::UDP, &self.last_ports.udp_ports) ]; for (proto, list) in actions.iter() { for port in *list { let service_str = format!("{}:{}", self.private_ip, port); let service = service_str.parse::().context("Invalid socket address")?; self.gateway.add_port(*proto, *port, service, self.expire.as_secs() as u32, "diplonat").await?; debug!("IGD request successful for {:#?} {}", proto, service); } } return Ok(()); } }