Nolocal close

Cargo.toml

@@ -84,6 +84,7 @@ async-global-executor = "2.4.1"
 async-io = "2.3.4"
 async-std = { version = "1.6.5", features = ["tokio1"] }
 async-trait = "0.1.82"
+async-channel = "2.3.1"
 base64 = "0.22.1"
 bincode = "1.3.3"
 bytes = "1.7.1"

zenoh/Cargo.toml

@@ -68,6 +68,7 @@ unstable = ["internal_config", "zenoh-keyexpr/unstable", "zenoh-config/unstable"
 internal_config = []
 
 [dependencies]
+async-channel = { workspace = true }
 tokio = { workspace = true, features = ["rt", "macros", "time"] }
 tokio-util = { workspace = true }
 ahash = { workspace = true }

zenoh/src/api/builders/close.rs

@@ -98,19 +98,13 @@ impl<TCloseable: Closeable> IntoFuture for CloseBuilder<TCloseable> {
 
 #[cfg(all(feature = "unstable", feature = "internal"))]
 /// A builder for close operations running in background
-// NOTE: `Closeable` is only pub(crate) because it is zenoh-internal trait, so we don't
-// care about the `private_bounds` lint in this particular case.
 #[doc(hidden)]
-#[allow(private_bounds)]
 pub struct BackgroundCloseBuilder<TOutput: Send + 'static> {
     inner: Pin<Box<dyn Future<Output = TOutput> + Send>>,
 }
 
 #[cfg(all(feature = "unstable", feature = "internal"))]
 #[doc(hidden)]
-// NOTE: `Closeable` is only pub(crate) because it is zenoh-internal trait, so we don't
-// care about the `private_bounds` lint in this particular case.
-#[allow(private_bounds)]
 impl<TOutput: Send + 'static> BackgroundCloseBuilder<TOutput> {
     fn new(inner: Pin<Box<dyn Future<Output = TOutput> + Send>>) -> Self {
         Self { inner }
@@ -119,7 +113,7 @@ impl<TOutput: Send + 'static> BackgroundCloseBuilder<TOutput> {
 
 #[cfg(all(feature = "unstable", feature = "internal"))]
 impl<TOutput: Send + 'static> Resolvable for BackgroundCloseBuilder<TOutput> {
-    type To = tokio::task::JoinHandle<TOutput>;
+    type To = NolocalJoinHandle<TOutput>;
 }
 
 #[cfg(all(feature = "unstable", feature = "internal"))]
@@ -134,10 +128,65 @@ impl<TOutput: Send + 'static> IntoFuture for BackgroundCloseBuilder<TOutput> {
     type Output = <Self as Resolvable>::To;
     type IntoFuture = Pin<Box<dyn Future<Output = <Self as IntoFuture>::Output> + Send>>;
 
-    // NOTE: yes, we need to return a future that returns JoinHandle
-    #[allow(clippy::async_yields_async)]
     fn into_future(self) -> Self::IntoFuture {
-        Box::pin(async move { ZRuntime::Application.spawn(self.inner) }.into_future())
+        Box::pin(
+            async move {
+                let (tx, rx) = async_channel::bounded::<TOutput>(1);
+
+                ZRuntime::Application.spawn(async move {
+                    tx.send(self.inner.await)
+                        .await
+                        .expect("BackgroundCloseBuilder: critical error sending the result")
+                });
+                NolocalJoinHandle::new(rx)
+            }
+            .into_future(),
+        )
+    }
+}
+
+#[cfg(all(feature = "unstable", feature = "internal"))]
+#[doc(hidden)]
+pub struct NolocalJoinHandle<TOutput: Send + 'static> {
+    rx: async_channel::Receiver<TOutput>,
+}
+
+#[cfg(all(feature = "unstable", feature = "internal"))]
+impl<TOutput: Send + 'static> NolocalJoinHandle<TOutput> {
+    fn new(rx: async_channel::Receiver<TOutput>) -> Self {
+        Self { rx }
+    }
+}
+
+#[cfg(all(feature = "unstable", feature = "internal"))]
+impl<TOutput: Send + 'static> Resolvable for NolocalJoinHandle<TOutput> {
+    type To = TOutput;
+}
+
+#[cfg(all(feature = "unstable", feature = "internal"))]
+impl<TOutput: Send + 'static> Wait for NolocalJoinHandle<TOutput> {
+    fn wait(self) -> Self::To {
+        self.rx
+            .recv_blocking()
+            .expect("NolocalJoinHandle: critical error receiving the result")
+    }
+}
+
+#[cfg(all(feature = "unstable", feature = "internal"))]
+impl<TOutput: Send + 'static> IntoFuture for NolocalJoinHandle<TOutput> {
+    type Output = <Self as Resolvable>::To;
+    type IntoFuture = Pin<Box<dyn Future<Output = <Self as IntoFuture>::Output> + Send>>;
+
+    fn into_future(self) -> Self::IntoFuture {
+        Box::pin(
+            async move {
+                self.rx
+                    .recv()
+                    .await
+                    .expect("NolocalJoinHandle: critical error receiving the result")
+            }
+            .into_future(),
+        )
     }
 }
 

zenoh/src/lib.rs

@@ -461,6 +461,12 @@ compile_error!(
 
 #[zenoh_macros::internal]
 pub mod internal {
+    #[zenoh_macros::unstable]
+    pub mod builders {
+        pub mod close {
+            pub use crate::api::builders::close::{BackgroundCloseBuilder, NolocalJoinHandle};
+        }
+    }
     pub mod traits {
         pub use crate::api::builders::sample::{
             EncodingBuilderTrait, QoSBuilderTrait, SampleBuilderTrait, TimestampBuilderTrait,

zenoh/tests/session.rs

@@ -425,8 +425,8 @@ async fn zenoh_session_close_in_background() {
     let close_task_2 = peer02.close().in_background().await;
 
     let close_all = async move {
-        close_task_1.await.unwrap().unwrap();
-        close_task_2.await.unwrap().unwrap();
+        close_task_1.await.unwrap();
+        close_task_2.await.unwrap();
     };
     ztimeout!(close_all);
 }