Function run_udp_forwarder

port-forward/src/udp.rs:27–125

Run a UDP port forwarder: listen on `listen_addr`, forward to `target`.

(
    listen_addr: SocketAddr,
    target: SocketAddr,
    cancel: CancellationToken,
)

/// Run a UDP port forwarder: listen on `listen_addr`, forward to `target`.
pub async fn run_udp_forwarder(
    listen_addr: SocketAddr,
    target: SocketAddr,
    cancel: CancellationToken,
) {
    let host_socket = match UdpSocket::bind(listen_addr).await {
        Ok(s) => s,
        Err(e) => {
            tracing::error!("udp bind {listen_addr} failed: {e}");
            return;
        }
    };
    tracing::info!("udp forwarding {listen_addr} -> {target}");

    let host_socket = Arc::new(host_socket);
    let clients: Arc<Mutex<HashMap<SocketAddr, ClientState>>> =
        Arc::new(Mutex::new(HashMap::new()));

    // Periodic cleanup of idle client entries
    let cleanup_clients = clients.clone();
    let cleanup_cancel = cancel.child_token();
    tokio::spawn(async move {
        loop {
            tokio::select! {
                _ = cleanup_cancel.cancelled() => break,
                _ = tokio::time::sleep(CLEANUP_INTERVAL) => {
                    let mut map = cleanup_clients.lock().await;
                    let now = Instant::now();
                    map.retain(|addr, entry| {
                        let alive = now.duration_since(entry.last_active) < UDP_IDLE_TIMEOUT;
                        if !alive {
                            tracing::debug!("udp client {addr} idle timeout");
                        }
                        alive
                        // Dropping ClientState cancels the return-path task via _cancel
                    });
                }
            }
        }
    });

    let mut buf = vec![0u8; UDP_BUF_SIZE];

    loop {
        tokio::select! {
            _ = cancel.cancelled() => break,
            result = host_socket.recv_from(&mut buf) => {
                let (n, client_addr) = match result {
                    Ok(v) => v,
                    Err(e) => {
                        tracing::warn!("udp recv on {listen_addr}: {e}");
                        continue;
                    }
                };

                let data = &buf[..n];
                let mut map = clients.lock().await;