feat: remove alive flag in session/undeclarable objects

zenoh/src/api/admin.rs

@@ -30,7 +30,7 @@ use super::{
     key_expr::KeyExpr,
     queryable::Query,
     sample::{DataInfo, Locality, SampleKind},
-    session::Session,
+    session::{Session, SessionClone},
 };
 
 macro_rules! ke_for_sure {
@@ -121,11 +121,11 @@ pub(crate) fn on_admin_query(session: &Session, query: Query) {
 
 #[derive(Clone)]
 pub(crate) struct Handler {
-    pub(crate) session: Arc<Session>,
+    pub(crate) session: Arc<SessionClone>,
 }
 
 impl Handler {
-    pub(crate) fn new(session: Session) -> Self {
+    pub(crate) fn new(session: SessionClone) -> Self {
         Self {
             session: Arc::new(session),
         }
@@ -193,7 +193,7 @@ impl TransportMulticastEventHandler for Handler {
 
 pub(crate) struct PeerHandler {
     pub(crate) expr: WireExpr<'static>,
-    pub(crate) session: Arc<Session>,
+    pub(crate) session: Arc<SessionClone>,
 }
 
 impl TransportPeerEventHandler for PeerHandler {

zenoh/src/api/liveliness.rs

@@ -15,6 +15,7 @@
 use std::{
     convert::TryInto,
     future::{IntoFuture, Ready},
+    mem::ManuallyDrop,
     sync::Arc,
     time::Duration,
 };
@@ -235,7 +236,6 @@ impl Wait for LivelinessTokenBuilder<'_, '_> {
             .map(|tok_state| LivelinessToken {
                 session,
                 state: tok_state,
-                alive: true,
             })
     }
 }
@@ -291,7 +291,6 @@ pub(crate) struct LivelinessTokenState {
 pub struct LivelinessToken<'a> {
     pub(crate) session: SessionRef<'a>,
     pub(crate) state: Arc<LivelinessTokenState>,
-    pub(crate) alive: bool,
 }
 
 /// A [`Resolvable`] returned when undeclaring a [`LivelinessToken`](LivelinessToken).
@@ -315,7 +314,9 @@ pub struct LivelinessToken<'a> {
 #[must_use = "Resolvables do nothing unless you resolve them using the `res` method from either `SyncResolve` or `AsyncResolve`"]
 #[zenoh_macros::unstable]
 pub struct LivelinessTokenUndeclaration<'a> {
-    token: LivelinessToken<'a>,
+    // ManuallyDrop wrapper prevents the drop code to be executed,
+    // which would lead to a double undeclaration
+    token: ManuallyDrop<LivelinessToken<'a>>,
 }
 
 #[zenoh_macros::unstable]
@@ -325,8 +326,7 @@ impl Resolvable for LivelinessTokenUndeclaration<'_> {
 
 #[zenoh_macros::unstable]
 impl Wait for LivelinessTokenUndeclaration<'_> {
-    fn wait(mut self) -> <Self as Resolvable>::To {
-        self.token.alive = false;
+    fn wait(self) -> <Self as Resolvable>::To {
         self.token.session.undeclare_liveliness(self.token.state.id)
     }
 }
@@ -374,16 +374,16 @@ impl<'a> LivelinessToken<'a> {
 #[zenoh_macros::unstable]
 impl<'a> Undeclarable<(), LivelinessTokenUndeclaration<'a>> for LivelinessToken<'a> {
     fn undeclare_inner(self, _: ()) -> LivelinessTokenUndeclaration<'a> {
-        LivelinessTokenUndeclaration { token: self }
+        LivelinessTokenUndeclaration {
+            token: ManuallyDrop::new(self),
+        }
     }
 }
 
 #[zenoh_macros::unstable]
 impl Drop for LivelinessToken<'_> {
     fn drop(&mut self) {
-        if self.alive {
-            let _ = self.session.undeclare_liveliness(self.state.id);
-        }
+        let _ = self.session.undeclare_liveliness(self.state.id);
     }
 }
 
@@ -553,7 +553,6 @@ where
                 subscriber: SubscriberInner {
                     session,
                     state: sub_state,
-                    alive: true,
                 },
                 handler,
             })

zenoh/src/api/publisher.rs

@@ -16,13 +16,13 @@ use std::{
     convert::TryFrom,
     fmt,
     future::{IntoFuture, Ready},
+    mem::ManuallyDrop,
     pin::Pin,
     task::{Context, Poll},
 };
 
 use futures::Sink;
 use zenoh_core::{zread, Resolvable, Resolve, Wait};
-use zenoh_keyexpr::keyexpr;
 use zenoh_protocol::{
     core::CongestionControl,
     network::{push::ext, Push},
@@ -442,7 +442,9 @@ impl PublisherDeclarations for std::sync::Arc<Publisher<'static>> {
 
 impl<'a> Undeclarable<(), PublisherUndeclaration<'a>> for Publisher<'a> {
     fn undeclare_inner(self, _: ()) -> PublisherUndeclaration<'a> {
-        PublisherUndeclaration { publisher: self }
+        PublisherUndeclaration {
+            publisher: ManuallyDrop::new(self),
+        }
     }
 }
 
@@ -461,21 +463,20 @@ impl<'a> Undeclarable<(), PublisherUndeclaration<'a>> for Publisher<'a> {
 /// ```
 #[must_use = "Resolvables do nothing unless you resolve them using the `res` method from either `SyncResolve` or `AsyncResolve`"]
 pub struct PublisherUndeclaration<'a> {
-    publisher: Publisher<'a>,
+    // ManuallyDrop wrapper prevents the drop code to be executed,
+    // which would lead to a double undeclaration
+    publisher: ManuallyDrop<Publisher<'a>>,
 }
 
 impl Resolvable for PublisherUndeclaration<'_> {
     type To = ZResult<()>;
 }
 
 impl Wait for PublisherUndeclaration<'_> {
-    fn wait(mut self) -> <Self as Resolvable>::To {
-        let Publisher {
-            session, id: eid, ..
-        } = &self.publisher;
-        session.undeclare_publisher_inner(*eid)?;
-        self.publisher.key_expr = unsafe { keyexpr::from_str_unchecked("") }.into();
-        Ok(())
+    fn wait(self) -> <Self as Resolvable>::To {
+        self.publisher
+            .session
+            .undeclare_publisher_inner(self.publisher.id)
     }
 }
 
@@ -490,9 +491,7 @@ impl IntoFuture for PublisherUndeclaration<'_> {
 
 impl Drop for Publisher<'_> {
     fn drop(&mut self) {
-        if !self.key_expr.is_empty() {
-            let _ = self.session.undeclare_publisher_inner(self.id);
-        }
+        let _ = self.session.undeclare_publisher_inner(self.id);
     }
 }
 
@@ -894,7 +893,6 @@ where
                 listener: MatchingListenerInner {
                     publisher: self.publisher,
                     state: listener_state,
-                    alive: true,
                 },
                 receiver,
             })
@@ -939,7 +937,6 @@ impl std::fmt::Debug for MatchingListenerState {
 pub(crate) struct MatchingListenerInner<'a> {
     pub(crate) publisher: PublisherRef<'a>,
     pub(crate) state: std::sync::Arc<MatchingListenerState>,
-    pub(crate) alive: bool,
 }
 
 #[zenoh_macros::unstable]
@@ -953,7 +950,9 @@ impl<'a> MatchingListenerInner<'a> {
 #[zenoh_macros::unstable]
 impl<'a> Undeclarable<(), MatchingListenerUndeclaration<'a>> for MatchingListenerInner<'a> {
     fn undeclare_inner(self, _: ()) -> MatchingListenerUndeclaration<'a> {
-        MatchingListenerUndeclaration { subscriber: self }
+        MatchingListenerUndeclaration {
+            subscriber: ManuallyDrop::new(self),
+        }
     }
 }
 
@@ -1033,7 +1032,9 @@ impl<Receiver> std::ops::DerefMut for MatchingListener<'_, Receiver> {
 
 #[zenoh_macros::unstable]
 pub struct MatchingListenerUndeclaration<'a> {
-    subscriber: MatchingListenerInner<'a>,
+    // ManuallyDrop wrapper prevents the drop code to be executed,
+    // which would lead to a double undeclaration
+    subscriber: ManuallyDrop<MatchingListenerInner<'a>>,
 }
 
 #[zenoh_macros::unstable]
@@ -1043,8 +1044,7 @@ impl Resolvable for MatchingListenerUndeclaration<'_> {
 
 #[zenoh_macros::unstable]
 impl Wait for MatchingListenerUndeclaration<'_> {
-    fn wait(mut self) -> <Self as Resolvable>::To {
-        self.subscriber.alive = false;
+    fn wait(self) -> <Self as Resolvable>::To {
         self.subscriber
             .publisher
             .session
@@ -1065,12 +1065,10 @@ impl IntoFuture for MatchingListenerUndeclaration<'_> {
 #[zenoh_macros::unstable]
 impl Drop for MatchingListenerInner<'_> {
     fn drop(&mut self) {
-        if self.alive {
-            let _ = self
-                .publisher
-                .session
-                .undeclare_matches_listener_inner(self.state.id);
-        }
+        let _ = self
+            .publisher
+            .session
+            .undeclare_matches_listener_inner(self.state.id);
     }
 }
 

zenoh/src/api/queryable.rs

@@ -14,6 +14,7 @@
 use std::{
     fmt,
     future::{IntoFuture, Ready},
+    mem::ManuallyDrop,
     ops::{Deref, DerefMut},
     sync::Arc,
 };
@@ -611,12 +612,13 @@ impl fmt::Debug for QueryableState {
 pub(crate) struct CallbackQueryable<'a> {
     pub(crate) session: SessionRef<'a>,
     pub(crate) state: Arc<QueryableState>,
-    pub(crate) alive: bool,
 }
 
 impl<'a> Undeclarable<(), QueryableUndeclaration<'a>> for CallbackQueryable<'a> {
     fn undeclare_inner(self, _: ()) -> QueryableUndeclaration<'a> {
-        QueryableUndeclaration { queryable: self }
+        QueryableUndeclaration {
+            queryable: ManuallyDrop::new(self),
+        }
     }
 }
 
@@ -635,16 +637,17 @@ impl<'a> Undeclarable<(), QueryableUndeclaration<'a>> for CallbackQueryable<'a>
 /// ```
 #[must_use = "Resolvables do nothing unless you resolve them using the `res` method from either `SyncResolve` or `AsyncResolve`"]
 pub struct QueryableUndeclaration<'a> {
-    queryable: CallbackQueryable<'a>,
+    // ManuallyDrop wrapper prevents the drop code to be executed,
+    // which would lead to a double undeclaration
+    queryable: ManuallyDrop<CallbackQueryable<'a>>,
 }
 
 impl Resolvable for QueryableUndeclaration<'_> {
     type To = ZResult<()>;
 }
 
 impl Wait for QueryableUndeclaration<'_> {
-    fn wait(mut self) -> <Self as Resolvable>::To {
-        self.queryable.alive = false;
+    fn wait(self) -> <Self as Resolvable>::To {
         self.queryable
             .session
             .close_queryable(self.queryable.state.id)
@@ -662,9 +665,7 @@ impl<'a> IntoFuture for QueryableUndeclaration<'a> {
 
 impl Drop for CallbackQueryable<'_> {
     fn drop(&mut self) {
-        if self.alive {
-            let _ = self.session.close_queryable(self.state.id);
-        }
+        let _ = self.session.close_queryable(self.state.id);
     }
 }
 
@@ -944,7 +945,6 @@ where
                 queryable: CallbackQueryable {
                     session,
                     state: qable_state,
-                    alive: true,
                 },
                 handler: receiver,
             })