diff --git a/io/zenoh-transport/src/unicast/lowlatency/transport.rs b/io/zenoh-transport/src/unicast/lowlatency/transport.rs index c9ed19c463..e4e9a9308b 100644 --- a/io/zenoh-transport/src/unicast/lowlatency/transport.rs +++ b/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 + } } diff --git a/io/zenoh-transport/src/unicast/mod.rs b/io/zenoh-transport/src/unicast/mod.rs index 1ef8865a16..45e5cf5d86 100644 --- a/io/zenoh-transport/src/unicast/mod.rs +++ b/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; diff --git a/io/zenoh-transport/src/unicast/test_helpers.rs b/io/zenoh-transport/src/unicast/test_helpers.rs index 5ba22eee07..6a84f4eace 100644 --- a/io/zenoh-transport/src/unicast/test_helpers.rs +++ b/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>, diff --git a/io/zenoh-transport/src/unicast/transport_unicast_inner.rs b/io/zenoh-transport/src/unicast/transport_unicast_inner.rs index 4f41a71209..4132bb18c5 100644 --- a/io/zenoh-transport/src/unicast/transport_unicast_inner.rs +++ 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>, diff --git a/io/zenoh-transport/src/unicast/universal/transport.rs b/io/zenoh-transport/src/unicast/universal/transport.rs index f9d1af5df4..597ccf8887 100644 --- a/io/zenoh-transport/src/unicast/universal/transport.rs +++ 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 { zread!(self.links) .get_links() diff --git a/io/zenoh-transport/tests/unicast_transport.rs b/io/zenoh-transport/tests/unicast_transport.rs index 446aafe5ad..07e0b4a2f2 100644 --- a/io/zenoh-transport/tests/unicast_transport.rs +++ b/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()); +}