feat: add close_link method to TransportUnicast API

io/zenoh-transport/src/unicast/lowlatency/transport.rs

@@ -330,4 +330,14 @@ impl TransportUnicastTrait for TransportUnicastLowlatency {
         tracing::trace!("Closing transport with peer: {}", self.config.zid);
         self.finalize(reason).await
     }
+
+    async fn close_link(&self, link: Link) -> ZResult<()> {
+        // Lowlatency transport has at most one link, so closing the link
+        // is equivalent to closing the entire transport.
+        tracing::trace!(
+            "Closing link {link} on lowlatency transport with peer: {}",
+            self.config.zid
+        );
+        self.finalize(close::reason::GENERIC).await
+    }
 }

io/zenoh-transport/src/unicast/mod.rs

@@ -152,6 +152,28 @@ impl TransportUnicast {
         }
     }
 
+    /// Close a specific link within this transport.
+    ///
+    /// If the transport has multiple links (multilink), only the specified
+    /// link is closed and the transport remains alive. If this is the last
+    /// link, the entire transport is closed.
+    ///
+    /// # Example
+    /// ```no_run
+    /// # async fn example(transport: zenoh_transport::unicast::TransportUnicast) {
+    /// // Close a specific link while keeping the transport alive
+    /// let links = transport.get_links().unwrap();
+    /// if links.len() > 1 {
+    ///     transport.close_link(links[0].clone()).await.unwrap();
+    /// }
+    /// # }
+    /// ```
+    #[inline(always)]
+    pub async fn close_link(&self, link: Link) -> ZResult<()> {
+        let transport = self.get_inner()?;
+        transport.close_link(link).await
+    }
+
     /// Returns the transport stats, or an error if the transport is closed.
     ///
     /// Warning: returning an error prevents interceptors to initialize;

io/zenoh-transport/src/unicast/test_helpers.rs

@@ -144,6 +144,10 @@ impl TransportUnicastTrait for MockTransportUnicastInner {
         Ok(())
     }
 
+    async fn close_link(&self, _link: Link) -> ZResult<()> {
+        Ok(())
+    }
+
     fn add_debug_fields<'a, 'b: 'a, 'c>(
         &self,
         s: &'c mut DebugStruct<'a, 'b>,

io/zenoh-transport/src/unicast/transport_unicast_inner.rs

@@ -105,6 +105,13 @@ pub(crate) trait TransportUnicastTrait: Send + Sync {
     /*************************************/
     async fn close(&self, reason: u8) -> ZResult<()>;
 
+    /// Close a specific link within this transport.
+    ///
+    /// If the transport has multiple links (multilink), only the specified
+    /// link is closed and the transport remains alive. If this is the last
+    /// link, the entire transport is closed.
+    async fn close_link(&self, link: Link) -> ZResult<()>;
+
     fn add_debug_fields<'a, 'b: 'a, 'c>(
         &self,
         s: &'c mut DebugStruct<'a, 'b>,

io/zenoh-transport/src/unicast/universal/transport.rs

@@ -423,6 +423,10 @@ impl TransportUnicastTrait for TransportUnicastUniversal {
         self.delete().await
     }
 
+    async fn close_link(&self, link: Link) -> ZResult<()> {
+        self.del_link(link).await
+    }
+
     fn get_links(&self) -> Vec<Link> {
         zread!(self.links)
             .get_links()

io/zenoh-transport/tests/unicast_transport.rs

@@ -3165,3 +3165,152 @@ async fn transport_unicast_with_zid_multilink() {
     ztimeout!(router_manager.close());
     ztimeout!(client_manager.close());
 }
+
+#[cfg(all(feature = "transport_tcp", feature = "transport_multilink"))]
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn transport_unicast_close_link_multilink() {
+    zenoh_util::init_log_from_env_or("error");
+
+    // Set up two endpoints on the router
+    let ep1: EndPoint = format!("tcp/127.0.0.1:{}", get_free_tcp_port())
+        .parse()
+        .unwrap();
+    let ep2: EndPoint = format!("tcp/127.0.0.1:{}", get_free_tcp_port())
+        .parse()
+        .unwrap();
+    let endpoints = vec![ep1.clone(), ep2.clone()];
+
+    // Open a multilink transport (max_links=2)
+    let (router_manager, _router_handler, client_manager, client_transport) =
+        open_transport_unicast(&endpoints, &endpoints, false, 2, 2).await;
+
+    // Verify we have 2 links
+    let links = client_transport.get_links().unwrap();
+    assert_eq!(links.len(), 2, "should have 2 links after multilink open");
+
+    // Close one link via close_link — transport should survive
+    let link_to_close = links[0].clone();
+    ztimeout!(client_transport.close_link(link_to_close)).unwrap();
+
+    // Give the transport time to process the link closure
+    tokio::time::sleep(SLEEP).await;
+
+    // Verify transport is still alive (not closed)
+    assert!(
+        client_transport.get_links().is_ok(),
+        "transport should still be alive after closing one of two links"
+    );
+
+    // Verify only 1 link remains
+    let remaining_links = client_transport.get_links().unwrap();
+    assert_eq!(
+        remaining_links.len(),
+        1,
+        "should have 1 remaining link after close_link"
+    );
+
+    // Close the last link — transport should be fully closed now
+    let last_link = remaining_links[0].clone();
+    ztimeout!(client_transport.close_link(last_link)).unwrap();
+
+    tokio::time::sleep(SLEEP).await;
+
+    // Verify transport is closed
+    assert!(
+        client_transport.get_links().is_err(),
+        "transport should be closed after last link removed"
+    );
+
+    // Clean up
+    ztimeout!(async {
+        while !router_manager.get_transports_unicast().await.is_empty() {
+            tokio::time::sleep(SLEEP).await;
+        }
+    });
+    for e in endpoints.iter() {
+        ztimeout!(router_manager.del_listener(e)).unwrap();
+    }
+    ztimeout!(router_manager.close());
+    ztimeout!(client_manager.close());
+}
+
+#[cfg(feature = "transport_tcp")]
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn transport_unicast_close_link_single_link() {
+    zenoh_util::init_log_from_env_or("error");
+
+    // Set up a single endpoint (no multilink)
+    let endpoint: EndPoint = format!("tcp/127.0.0.1:{}", get_free_tcp_port())
+        .parse()
+        .unwrap();
+    let endpoints = vec![endpoint.clone()];
+
+    let (router_manager, _router_handler, client_manager, client_transport) =
+        open_transport_unicast(&endpoints, &endpoints, false, 1, 1).await;
+
+    // Verify we have 1 link
+    let links = client_transport.get_links().unwrap();
+    assert_eq!(links.len(), 1);
+
+    // Close the only link — should close the entire transport
+    ztimeout!(client_transport.close_link(links[0].clone())).unwrap();
+
+    tokio::time::sleep(SLEEP).await;
+
+    // Verify transport is closed
+    assert!(
+        client_transport.get_links().is_err(),
+        "transport should be closed after closing the only link"
+    );
+
+    // Clean up
+    ztimeout!(async {
+        while !router_manager.get_transports_unicast().await.is_empty() {
+            tokio::time::sleep(SLEEP).await;
+        }
+    });
+    ztimeout!(router_manager.del_listener(&endpoint)).unwrap();
+    ztimeout!(router_manager.close());
+    ztimeout!(client_manager.close());
+}
+
+#[cfg(feature = "transport_tcp")]
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn transport_unicast_close_link_lowlatency() {
+    zenoh_util::init_log_from_env_or("error");
+
+    // Set up a single endpoint with lowlatency transport
+    let endpoint: EndPoint = format!("tcp/127.0.0.1:{}", get_free_tcp_port())
+        .parse()
+        .unwrap();
+    let endpoints = vec![endpoint.clone()];
+
+    let (router_manager, _router_handler, client_manager, client_transport) =
+        open_transport_unicast(&endpoints, &endpoints, true, 1, 1).await;
+
+    // Verify we have 1 link
+    let links = client_transport.get_links().unwrap();
+    assert_eq!(links.len(), 1);
+
+    // Close the only link via close_link — should close the entire transport
+    // (lowlatency transport has at most one link)
+    ztimeout!(client_transport.close_link(links[0].clone())).unwrap();
+
+    tokio::time::sleep(SLEEP).await;
+
+    // Verify transport is closed
+    assert!(
+        client_transport.get_links().is_err(),
+        "lowlatency transport should be closed after closing the only link"
+    );
+
+    // Clean up
+    ztimeout!(async {
+        while !router_manager.get_transports_unicast().await.is_empty() {
+            tokio::time::sleep(SLEEP).await;
+        }
+    });
+    ztimeout!(router_manager.del_listener(&endpoint)).unwrap();
+    ztimeout!(router_manager.close());
+    ztimeout!(client_manager.close());
+}