use std::net::SocketAddrV4; use anyhow::{Context, Result}; use igd::aio::*; use igd::PortMappingProtocol; use log::*; use tokio::{ select, sync::watch, time::{self, Duration}, }; 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(()); } }