Fix RAII for subscription, client, and service

Signed-off-by: Michael X. Grey <[email protected]>

Author: mxgrey

rclrs/src/client.rs

@@ -11,7 +11,7 @@ use rosidl_runtime_rs::Message;
 use crate::{
     error::{RclReturnCode, ToResult},
     rcl_bindings::*,
-    MessageCow, NodeHandle, RclrsError, ENTITY_LIFECYCLE_MUTEX,
+    MessageCow, Node, NodeHandle, RclrsError, ENTITY_LIFECYCLE_MUTEX,
 };
 
 // SAFETY: The functions accessing this type, including drop(), shouldn't care about the thread
@@ -76,14 +76,17 @@ where
     pub(crate) handle: Arc<ClientHandle>,
     requests: Mutex<HashMap<RequestId, RequestValue<T::Response>>>,
     futures: Arc<Mutex<HashMap<RequestId, oneshot::Sender<T::Response>>>>,
+    /// Ensure the parent node remains alive as long as the subscription is held.
+    /// This implementation will change in the future.
+    node: Arc<Node>,
 }
 
 impl<T> Client<T>
 where
     T: rosidl_runtime_rs::Service,
 {
     /// Creates a new client.
-    pub(crate) fn new(node_handle: Arc<NodeHandle>, topic: &str) -> Result<Self, RclrsError>
+    pub(crate) fn new(node: &Arc<Node>, topic: &str) -> Result<Self, RclrsError>
     // This uses pub(crate) visibility to avoid instantiating this struct outside
     // [`Node::create_client`], see the struct's documentation for the rationale
     where
@@ -102,7 +105,7 @@ where
         let client_options = unsafe { rcl_client_get_default_options() };
 
         {
-            let rcl_node = node_handle.rcl_node.lock().unwrap();
+            let rcl_node = node.handle.rcl_node.lock().unwrap();
             let _lifecycle_lock = ENTITY_LIFECYCLE_MUTEX.lock().unwrap();
 
             // SAFETY:
@@ -136,6 +139,7 @@ where
             futures: Arc::new(Mutex::new(
                 HashMap::<RequestId, oneshot::Sender<T::Response>>::new(),
             )),
+            node: Arc::clone(node),
         })
     }
 

rclrs/src/service.rs

@@ -9,7 +9,7 @@ use rosidl_runtime_rs::Message;
 use crate::{
     error::{RclReturnCode, ToResult},
     rcl_bindings::*,
-    MessageCow, NodeHandle, RclrsError, ENTITY_LIFECYCLE_MUTEX,
+    MessageCow, Node, NodeHandle, RclrsError, ENTITY_LIFECYCLE_MUTEX,
 };
 
 // SAFETY: The functions accessing this type, including drop(), shouldn't care about the thread
@@ -73,6 +73,9 @@ where
     pub(crate) handle: Arc<ServiceHandle>,
     /// The callback function that runs when a request was received.
     pub callback: Mutex<ServiceCallback<T::Request, T::Response>>,
+    /// Ensure the parent node remains alive as long as the subscription is held.
+    /// This implementation will change in the future.
+    node: Arc<Node>,
 }
 
 impl<T> Service<T>
@@ -81,7 +84,7 @@ where
 {
     /// Creates a new service.
     pub(crate) fn new<F>(
-        node_handle: Arc<NodeHandle>,
+        node: &Arc<Node>,
         topic: &str,
         callback: F,
     ) -> Result<Self, RclrsError>
@@ -104,7 +107,7 @@ where
         let service_options = unsafe { rcl_service_get_default_options() };
 
         {
-            let rcl_node = node_handle.rcl_node.lock().unwrap();
+            let rcl_node = node.handle.rcl_node.lock().unwrap();
             let _lifecycle_lock = ENTITY_LIFECYCLE_MUTEX.lock().unwrap();
             unsafe {
                 // SAFETY:
@@ -133,6 +136,7 @@ where
 
         Ok(Self {
             handle,
+            node: Arc::clone(node),
             callback: Mutex::new(Box::new(callback)),
         })
     }

rclrs/src/subscription.rs

@@ -10,7 +10,7 @@ use crate::{
     error::{RclReturnCode, ToResult},
     qos::QoSProfile,
     rcl_bindings::*,
-    NodeHandle, RclrsError, ENTITY_LIFECYCLE_MUTEX,
+    Node, NodeHandle, RclrsError, ENTITY_LIFECYCLE_MUTEX,
 };
 
 mod callback;
@@ -31,7 +31,7 @@ unsafe impl Send for rcl_subscription_t {}
 /// [1]: <https://doc.rust-lang.org/reference/destructors.html>
 pub struct SubscriptionHandle {
     rcl_subscription: Mutex<rcl_subscription_t>,
-    node_handle: Arc<NodeHandle>,
+    node_handle: Arc<Node>,
     pub(crate) in_use_by_wait_set: Arc<AtomicBool>,
 }
 
@@ -84,6 +84,9 @@ where
     pub(crate) handle: Arc<SubscriptionHandle>,
     /// The callback function that runs when a message was received.
     pub callback: Mutex<AnySubscriptionCallback<T>>,
+    /// Ensure the parent node remains alive as long as the subscription is held.
+    /// This implementation will change in the future.
+    node: Arc<Node>,
     message: PhantomData<T>,
 }
 
@@ -93,7 +96,7 @@ where
 {
     /// Creates a new subscription.
     pub(crate) fn new<Args>(
-        node_handle: Arc<NodeHandle>,
+        node: &Arc<Node>,
         topic: &str,
         qos: QoSProfile,
         callback: impl SubscriptionCallback<T, Args>,
@@ -117,7 +120,7 @@ where
         subscription_options.qos = qos.into();
 
         {
-            let rcl_node = node_handle.rcl_node.lock().unwrap();
+            let rcl_node = node.handle.rcl_node.lock().unwrap();
             let _lifecycle_lock = ENTITY_LIFECYCLE_MUTEX.lock().unwrap();
             unsafe {
                 // SAFETY:
@@ -146,6 +149,7 @@ where
         Ok(Self {
             handle,
             callback: Mutex::new(callback.into_callback()),
+            node: Arc::clone(node),
             message: PhantomData,
         })
     }
@@ -396,4 +400,36 @@ mod tests {
         );
         Ok(())
     }
+
+    #[test]
+    fn test_node_subscription_raii() {
+        use rclrs::*;
+        use std::sync::atomic::Ordering;
+
+        let executor = Context::default().create_basic_executor();
+
+        let triggered = Arc::new(AtomicBool::new(false));
+        let callback = |_| {
+            triggered.store(true, Ordering::AcqRel);
+        };
+
+        let (subscription, publisher) = {
+            let node = executor
+                .create_node(&format!("test_node_subscription_raii_{}", line!()))
+                .unwrap();
+
+            let qos = QoSProfile::default().keep_all().reliable();
+            let subscription = node.create_subscription::<msg::Empty>("test_topic", qos, callback).unwrap();
+            let publisher = node.create_publisher::<msg::Empty>("test_topic", qos).unwrap();
+
+            (subscription, publisher)
+        };
+
+        publisher.publish(msg::Empty {});
+        let start_time = std::time::Instant::now();
+        while !triggered.load(Ordering::AcqRel) {
+            assert!(executor.spin(SpinOptions::spin_once()).is_empty());
+            assert!(start_time.elapsed() < std::time::Duration::from_secs(10));
+        }
+    }
 }