Make constructor methods for pub/sub/etc. take rcl_node_t mutex (#290)

The constructors are not public, so we're free to use internal types as arguments.

This signature makes it possible, or at least easier, to write parameter services that are owned by
the node: With this change, the Node does not need to be constructed before the parameter services are constructed,
which would somewhat contradict the node owning these services.

Author: nnmm

rclrs/src/client.rs

@@ -9,7 +9,6 @@ use rosidl_runtime_rs::Message;
 
 use crate::error::{RclReturnCode, ToResult};
 use crate::MessageCow;
-use crate::Node;
 use crate::{rcl_bindings::*, RclrsError};
 
 // SAFETY: The functions accessing this type, including drop(), shouldn't care about the thread
@@ -56,8 +55,11 @@ type RequestId = i64;
 
 /// Main class responsible for sending requests to a ROS service.
 ///
-/// The only available way to instantiate clients is via [`Node::create_client`], this is to
-/// ensure that [`Node`]s can track all the clients that have been created.
+/// The only available way to instantiate clients is via [`Node::create_client`][1], this is to
+/// ensure that [`Node`][2]s can track all the clients that have been created.
+///
+/// [1]: crate::Node::create_client
+/// [2]: crate::Node
 pub struct Client<T>
 where
     T: rosidl_runtime_rs::Service,
@@ -72,7 +74,7 @@ where
     T: rosidl_runtime_rs::Service,
 {
     /// Creates a new client.
-    pub(crate) fn new(node: &Node, topic: &str) -> Result<Self, RclrsError>
+    pub(crate) fn new(rcl_node_mtx: Arc<Mutex<rcl_node_t>>, 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
@@ -86,7 +88,6 @@ where
             err,
             s: topic.into(),
         })?;
-        let rcl_node = { &mut *node.rcl_node_mtx.lock().unwrap() };
 
         // SAFETY: No preconditions for this function.
         let client_options = unsafe { rcl_client_get_default_options() };
@@ -98,7 +99,7 @@ where
             // afterwards.
             rcl_client_init(
                 &mut rcl_client,
-                rcl_node,
+                &*rcl_node_mtx.lock().unwrap(),
                 type_support,
                 topic_c_string.as_ptr(),
                 &client_options,
@@ -108,7 +109,7 @@ where
 
         let handle = Arc::new(ClientHandle {
             rcl_client_mtx: Mutex::new(rcl_client),
-            rcl_node_mtx: node.rcl_node_mtx.clone(),
+            rcl_node_mtx,
             in_use_by_wait_set: Arc::new(AtomicBool::new(false)),
         });
 

rclrs/src/node.rs

@@ -184,7 +184,7 @@ impl Node {
     where
         T: rosidl_runtime_rs::Service,
     {
-        let client = Arc::new(Client::<T>::new(self, topic)?);
+        let client = Arc::new(Client::<T>::new(Arc::clone(&self.rcl_node_mtx), topic)?);
         self.clients
             .push(Arc::downgrade(&client) as Weak<dyn ClientBase>);
         Ok(client)
@@ -243,7 +243,7 @@ impl Node {
     where
         T: Message,
     {
-        Publisher::<T>::new(self, topic, qos)
+        Publisher::<T>::new(Arc::clone(&self.rcl_node_mtx), topic, qos)
     }
 
     /// Creates a [`Service`][1].
@@ -259,7 +259,11 @@ impl Node {
         T: rosidl_runtime_rs::Service,
         F: Fn(&rmw_request_id_t, T::Request) -> T::Response + 'static + Send,
     {
-        let service = Arc::new(Service::<T>::new(self, topic, callback)?);
+        let service = Arc::new(Service::<T>::new(
+            Arc::clone(&self.rcl_node_mtx),
+            topic,
+            callback,
+        )?);
         self.services
             .push(Arc::downgrade(&service) as Weak<dyn ServiceBase>);
         Ok(service)
@@ -278,7 +282,12 @@ impl Node {
     where
         T: Message,
     {
-        let subscription = Arc::new(Subscription::<T>::new(self, topic, qos, callback)?);
+        let subscription = Arc::new(Subscription::<T>::new(
+            Arc::clone(&self.rcl_node_mtx),
+            topic,
+            qos,
+            callback,
+        )?);
         self.subscriptions
             .push(Arc::downgrade(&subscription) as Weak<dyn SubscriptionBase>);
         Ok(subscription)

rclrs/src/publisher.rs

@@ -9,7 +9,6 @@ use rosidl_runtime_rs::{Message, RmwMessage};
 use crate::error::{RclrsError, ToResult};
 use crate::qos::QoSProfile;
 use crate::rcl_bindings::*;
-use crate::Node;
 
 mod loaned_message;
 pub use loaned_message::*;
@@ -69,7 +68,11 @@ where
     /// Creates a new `Publisher`.
     ///
     /// Node and namespace changes are always applied _before_ topic remapping.
-    pub fn new(node: &Node, topic: &str, qos: QoSProfile) -> Result<Self, RclrsError>
+    pub fn new(
+        rcl_node_mtx: Arc<Mutex<rcl_node_t>>,
+        topic: &str,
+        qos: QoSProfile,
+    ) -> Result<Self, RclrsError>
     where
         T: Message,
     {
@@ -81,7 +84,6 @@ where
             err,
             s: topic.into(),
         })?;
-        let rcl_node = &mut *node.rcl_node_mtx.lock().unwrap();
 
         // SAFETY: No preconditions for this function.
         let mut publisher_options = unsafe { rcl_publisher_get_default_options() };
@@ -94,7 +96,7 @@ where
             // TODO: type support?
             rcl_publisher_init(
                 &mut rcl_publisher,
-                rcl_node,
+                &*rcl_node_mtx.lock().unwrap(),
                 type_support_ptr,
                 topic_c_string.as_ptr(),
                 &publisher_options,
@@ -104,7 +106,7 @@ where
 
         Ok(Self {
             rcl_publisher_mtx: Mutex::new(rcl_publisher),
-            rcl_node_mtx: Arc::clone(&node.rcl_node_mtx),
+            rcl_node_mtx,
             type_support_ptr,
             message: PhantomData,
         })

rclrs/src/service.rs

@@ -6,7 +6,7 @@ use std::sync::{Arc, Mutex, MutexGuard};
 use rosidl_runtime_rs::Message;
 
 use crate::error::{RclReturnCode, ToResult};
-use crate::{rcl_bindings::*, MessageCow, Node, RclrsError};
+use crate::{rcl_bindings::*, MessageCow, RclrsError};
 
 // SAFETY: The functions accessing this type, including drop(), shouldn't care about the thread
 // they are running in. Therefore, this type can be safely sent to another thread.
@@ -51,8 +51,11 @@ type ServiceCallback<Request, Response> =
 
 /// Main class responsible for responding to requests sent by ROS clients.
 ///
-/// The only available way to instantiate services is via [`Node::create_service`], this is to
-/// ensure that [`Node`]s can track all the services that have been created.
+/// The only available way to instantiate services is via [`Node::create_service()`][1], this is to
+/// ensure that [`Node`][2]s can track all the services that have been created.
+///
+/// [1]: crate::Node::create_service
+/// [2]: crate::Node
 pub struct Service<T>
 where
     T: rosidl_runtime_rs::Service,
@@ -67,7 +70,11 @@ where
     T: rosidl_runtime_rs::Service,
 {
     /// Creates a new service.
-    pub(crate) fn new<F>(node: &Node, topic: &str, callback: F) -> Result<Self, RclrsError>
+    pub(crate) fn new<F>(
+        rcl_node_mtx: Arc<Mutex<rcl_node_t>>,
+        topic: &str,
+        callback: F,
+    ) -> Result<Self, RclrsError>
     // This uses pub(crate) visibility to avoid instantiating this struct outside
     // [`Node::create_service`], see the struct's documentation for the rationale
     where
@@ -82,7 +89,6 @@ where
             err,
             s: topic.into(),
         })?;
-        let rcl_node = &mut *node.rcl_node_mtx.lock().unwrap();
 
         // SAFETY: No preconditions for this function.
         let service_options = unsafe { rcl_service_get_default_options() };
@@ -93,8 +99,8 @@ where
             // The topic name and the options are copied by this function, so they can be dropped
             // afterwards.
             rcl_service_init(
-                &mut rcl_service as *mut _,
-                rcl_node as *mut _,
+                &mut rcl_service,
+                &*rcl_node_mtx.lock().unwrap(),
                 type_support,
                 topic_c_string.as_ptr(),
                 &service_options as *const _,
@@ -104,7 +110,7 @@ where
 
         let handle = Arc::new(ServiceHandle {
             rcl_service_mtx: Mutex::new(rcl_service),
-            rcl_node_mtx: node.rcl_node_mtx.clone(),
+            rcl_node_mtx,
             in_use_by_wait_set: Arc::new(AtomicBool::new(false)),
         });
 

rclrs/src/subscription.rs

@@ -8,7 +8,6 @@ use rosidl_runtime_rs::{Message, RmwMessage};
 
 use crate::error::{RclReturnCode, ToResult};
 use crate::qos::QoSProfile;
-use crate::Node;
 use crate::{rcl_bindings::*, RclrsError};
 
 mod callback;
@@ -63,11 +62,13 @@ pub trait SubscriptionBase: Send + Sync {
 /// When a subscription is created, it may take some time to get "matched" with a corresponding
 /// publisher.
 ///
-/// The only available way to instantiate subscriptions is via [`Node::create_subscription`], this
-/// is to ensure that [`Node`]s can track all the subscriptions that have been created.
+/// The only available way to instantiate subscriptions is via [`Node::create_subscription()`][3], this
+/// is to ensure that [`Node`][4]s can track all the subscriptions that have been created.
 ///
 /// [1]: crate::spin_once
 /// [2]: crate::spin
+/// [3]: crate::Node::create_subscription
+/// [4]: crate::Node
 pub struct Subscription<T>
 where
     T: Message,
@@ -84,7 +85,7 @@ where
 {
     /// Creates a new subscription.
     pub(crate) fn new<Args>(
-        node: &Node,
+        rcl_node_mtx: Arc<Mutex<rcl_node_t>>,
         topic: &str,
         qos: QoSProfile,
         callback: impl SubscriptionCallback<T, Args>,
@@ -102,7 +103,6 @@ where
             err,
             s: topic.into(),
         })?;
-        let rcl_node = &mut *node.rcl_node_mtx.lock().unwrap();
 
         // SAFETY: No preconditions for this function.
         let mut subscription_options = unsafe { rcl_subscription_get_default_options() };
@@ -115,7 +115,7 @@ where
             // TODO: type support?
             rcl_subscription_init(
                 &mut rcl_subscription,
-                rcl_node,
+                &*rcl_node_mtx.lock().unwrap(),
                 type_support,
                 topic_c_string.as_ptr(),
                 &subscription_options,
@@ -125,7 +125,7 @@ where
 
         let handle = Arc::new(SubscriptionHandle {
             rcl_subscription_mtx: Mutex::new(rcl_subscription),
-            rcl_node_mtx: node.rcl_node_mtx.clone(),
+            rcl_node_mtx,
             in_use_by_wait_set: Arc::new(AtomicBool::new(false)),
         });