Always add subscription/client/service to the node's list of waitables

Author: nnmm

rclrs/src/node.rs

@@ -184,17 +184,11 @@ impl Node {
     ///
     /// [1]: crate::Client
     // TODO: make client's lifetime depend on node's lifetime
-    pub fn create_client<T>(
-        &mut self,
-        topic: &str,
-    ) -> Result<Arc<crate::node::client::Client<T>>, RclrsError>
+    pub fn create_client<T>(&mut self, topic: &str) -> Result<Arc<Client<T>>, RclrsError>
     where
         T: rosidl_runtime_rs::Service,
     {
-        let client = Arc::new(crate::node::client::Client::<T>::new(self, topic)?);
-        self.clients
-            .push(Arc::downgrade(&client) as Weak<dyn ClientBase>);
-        Ok(client)
+        Client::<T>::new(self, topic)
     }
 
     /// Creates a [`Publisher`][1].
@@ -220,17 +214,12 @@ impl Node {
         &mut self,
         topic: &str,
         callback: F,
-    ) -> Result<Arc<crate::node::service::Service<T>>, RclrsError>
+    ) -> Result<Arc<Service<T>>, RclrsError>
     where
         T: rosidl_runtime_rs::Service,
         F: Fn(&rmw_request_id_t, T::Request) -> T::Response + 'static + Send,
     {
-        let service = Arc::new(crate::node::service::Service::<T>::new(
-            self, topic, callback,
-        )?);
-        self.services
-            .push(Arc::downgrade(&service) as Weak<dyn ServiceBase>);
-        Ok(service)
+        Service::<T>::new(self, topic, callback)
     }
 
     /// Creates a [`Subscription`][1].
@@ -247,10 +236,7 @@ impl Node {
         T: Message,
         F: FnMut(T) + 'static + Send,
     {
-        let subscription = Arc::new(Subscription::<T>::new(self, topic, qos, callback)?);
-        self.subscriptions
-            .push(Arc::downgrade(&subscription) as Weak<dyn SubscriptionBase>);
-        Ok(subscription)
+        Subscription::new(self, topic, qos, callback)
     }
 
     /// Returns the subscriptions that have not been dropped yet.

rclrs/src/node/client.rs

@@ -78,7 +78,7 @@ where
     T: rosidl_runtime_rs::Service,
 {
     /// Creates a new client.
-    pub fn new(node: &Node, topic: &str) -> Result<Self, RclrsError>
+    pub fn new(node: &mut Node, topic: &str) -> Result<Arc<Self>, RclrsError>
     where
         T: rosidl_runtime_rs::Service,
     {
@@ -114,14 +114,15 @@ where
             rcl_client_mtx: Mutex::new(rcl_client),
             rcl_node_mtx: node.rcl_node_mtx.clone(),
         });
-
-        Ok(Self {
+        let client = Arc::new(Self {
             handle,
             requests: Mutex::new(HashMap::new()),
             futures: Arc::new(Mutex::new(
                 HashMap::<RequestId, oneshot::Sender<T::Response>>::new(),
             )),
-        })
+        });
+        node.clients.push(Arc::downgrade(&client) as _);
+        Ok(client)
     }
 
     /// Sends a request with a callback to be called with the response.

rclrs/src/node/service.rs

@@ -67,7 +67,7 @@ where
     T: rosidl_runtime_rs::Service,
 {
     /// Creates a new service.
-    pub fn new<F>(node: &Node, topic: &str, callback: F) -> Result<Self, RclrsError>
+    pub fn new<F>(node: &mut Node, topic: &str, callback: F) -> Result<Arc<Self>, RclrsError>
     where
         T: rosidl_runtime_rs::Service,
         F: Fn(&rmw_request_id_t, T::Request) -> T::Response + 'static + Send,
@@ -104,11 +104,13 @@ where
             handle: Mutex::new(service_handle),
             node_handle: node.rcl_node_mtx.clone(),
         });
-
-        Ok(Self {
+        let service = Arc::new(Self {
             handle,
             callback: Mutex::new(Box::new(callback)),
-        })
+        });
+        node.services.push(Arc::downgrade(&service) as _);
+
+        Ok(service)
     }
 
     /// Fetches a new request.

rclrs/src/node/subscription.rs

@@ -75,11 +75,11 @@ where
 {
     /// Creates a new subscription.
     pub fn new<F>(
-        node: &Node,
+        node: &mut Node,
         topic: &str,
         qos: QoSProfile,
         callback: F,
-    ) -> Result<Self, RclrsError>
+    ) -> Result<Arc<Self>, RclrsError>
     where
         T: Message,
         F: FnMut(T) + 'static + Send,
@@ -117,12 +117,13 @@ where
             rcl_subscription_mtx: Mutex::new(rcl_subscription),
             rcl_node_mtx: node.rcl_node_mtx.clone(),
         });
-
-        Ok(Self {
+        let subscription = Arc::new(Self {
             handle,
             callback: Mutex::new(Box::new(callback)),
             message: PhantomData,
-        })
+        });
+        node.subscriptions.push(Arc::downgrade(&subscription) as _);
+        Ok(subscription)
     }
 
     /// Returns the topic name of the subscription.
@@ -216,9 +217,9 @@ mod tests {
     fn test_instantiate_subscriber() -> Result<(), RclrsError> {
         let context =
             Context::new(vec![]).expect("Context instantiation is expected to be a success");
-        let node = create_node(&context, "test_new_subscriber")?;
+        let mut node = create_node(&context, "test_new_subscriber")?;
         let _subscriber = Subscription::<std_msgs::msg::String>::new(
-            &node,
+            &mut node,
             "test",
             QOS_PROFILE_DEFAULT,
             move |_: std_msgs::msg::String| {},