Expose qos profiles for service clients and servers. (#95)

Adds a QosProfile parameter for `Node::create_client` and `Node::node.create_service`. `QosProfile::default()` can be passed when updating existing code.
sequenceplanner · May 17, 2024 · a02a78f · a02a78f
1 parent e2d6aba
commit a02a78f
Show file tree

Hide file tree

Showing 9 changed files with 38 additions and 32 deletions.
diff --git a/r2r/examples/client.rs b/r2r/examples/client.rs
@@ -2,7 +2,7 @@ use futures::{executor::LocalPool, task::LocalSpawnExt, Future};
 
 use std::io::Write;
 
-use r2r::example_interfaces::srv::AddTwoInts;
+use r2r::{example_interfaces::srv::AddTwoInts, QosProfile};
 
 async fn requester_task(
     node_available: impl Future<Output = r2r::Result<()>>, c: r2r::Client<AddTwoInts::Service>,
@@ -29,7 +29,7 @@ async fn requester_task(
 fn main() -> Result<(), Box<dyn std::error::Error>> {
     let ctx = r2r::Context::create()?;
     let mut node = r2r::Node::create(ctx, "testnode", "")?;
-    let client = node.create_client::<AddTwoInts::Service>("/add_two_ints")?;
+    let client = node.create_client::<AddTwoInts::Service>("/add_two_ints", QosProfile::default())?;
 
     let service_available = r2r::Node::is_available(&client)?;
 

diff --git a/r2r/examples/service.rs b/r2r/examples/service.rs
@@ -1,6 +1,6 @@
 use futures::{executor::LocalPool, select, stream::StreamExt, task::LocalSpawnExt, FutureExt};
 
-use r2r::example_interfaces::srv::AddTwoInts;
+use r2r::{example_interfaces::srv::AddTwoInts, QosProfile};
 
 ///
 /// This example demonstrates how we can chain async service calls.
@@ -11,9 +11,9 @@ use r2r::example_interfaces::srv::AddTwoInts;
 fn main() -> Result<(), Box<dyn std::error::Error>> {
     let ctx = r2r::Context::create()?;
     let mut node = r2r::Node::create(ctx, "testnode", "")?;
-    let mut service = node.create_service::<AddTwoInts::Service>("/add_two_ints")?;
-    let service_delayed = node.create_service::<AddTwoInts::Service>("/add_two_ints_delayed")?;
-    let client = node.create_client::<AddTwoInts::Service>("/add_two_ints_delayed")?;
+    let mut service = node.create_service::<AddTwoInts::Service>("/add_two_ints", QosProfile::default())?;
+    let service_delayed = node.create_service::<AddTwoInts::Service>("/add_two_ints_delayed", QosProfile::default())?;
+    let client = node.create_client::<AddTwoInts::Service>("/add_two_ints_delayed", QosProfile::default())?;
     let mut timer = node.create_wall_timer(std::time::Duration::from_millis(250))?;
     let mut timer2 = node.create_wall_timer(std::time::Duration::from_millis(2000))?;
     // wait for service to be available

diff --git a/r2r/examples/tokio_client.rs b/r2r/examples/tokio_client.rs
@@ -1,11 +1,13 @@
+use r2r::QosProfile;
+
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let ctx = r2r::Context::create()?;
     let mut node = r2r::Node::create(ctx, "testnode", "")?;
     let duration = std::time::Duration::from_millis(2500);
 
     use r2r::example_interfaces::srv::AddTwoInts;
-    let client = node.create_client::<AddTwoInts::Service>("/add_two_ints")?;
+    let client = node.create_client::<AddTwoInts::Service>("/add_two_ints", QosProfile::default())?;
     let mut timer = node.create_wall_timer(duration)?;
     let waiting = r2r::Node::is_available(&client)?;
 

diff --git a/r2r/examples/tokio_examples.rs b/r2r/examples/tokio_examples.rs
@@ -75,7 +75,7 @@ async fn client(arc_node: Arc<Mutex<r2r::Node>>) -> Result<(), r2r::Error> {
     let (client, mut timer, service_available) = {
         // Limiting the scope when locking the arc
         let mut node = arc_node.lock().unwrap();
-        let client = node.create_client::<AddTwoInts::Service>("/add_two_ints")?;
+        let client = node.create_client::<AddTwoInts::Service>("/add_two_ints", QosProfile::default())?;
         let timer = node.create_wall_timer(std::time::Duration::from_secs(2))?;
         let service_available = r2r::Node::is_available(&client)?;
         (client, timer, service_available)
@@ -98,7 +98,7 @@ async fn service(arc_node: Arc<Mutex<r2r::Node>>) -> Result<(), r2r::Error> {
     let mut service = {
         // Limiting the scope when locking the arc
         let mut node = arc_node.lock().unwrap();
-        node.create_service::<AddTwoInts::Service>("/add_two_ints")?
+        node.create_service::<AddTwoInts::Service>("/add_two_ints", QosProfile::default())?
     };
     loop {
         match service.next().await {

diff --git a/r2r/examples/tokio_service.rs b/r2r/examples/tokio_service.rs
@@ -1,12 +1,13 @@
 use futures::stream::StreamExt;
+use r2r::QosProfile;
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let ctx = r2r::Context::create()?;
     let mut node = r2r::Node::create(ctx, "testnode", "")?;
 
     use r2r::example_interfaces::srv::AddTwoInts;
-    let mut service = node.create_service::<AddTwoInts::Service>("/add_two_ints")?;
+    let mut service = node.create_service::<AddTwoInts::Service>("/add_two_ints", QosProfile::default())?;
 
     let handle = tokio::task::spawn_blocking(move || loop {
         node.spin_once(std::time::Duration::from_millis(100));

diff --git a/r2r/examples/untyped_client.rs b/r2r/examples/untyped_client.rs
@@ -1,4 +1,5 @@
 use futures::{executor::LocalPool, task::LocalSpawnExt, Future};
+use r2r::QosProfile;
 
 async fn requester_task(
     node_available: impl Future<Output = r2r::Result<()>>, c: r2r::ClientUntyped,
@@ -25,7 +26,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     let ctx = r2r::Context::create()?;
     let mut node = r2r::Node::create(ctx, "testnode", "")?;
     let client =
-        node.create_client_untyped("/add_two_ints", "example_interfaces/srv/AddTwoInts")?;
+        node.create_client_untyped("/add_two_ints", "example_interfaces/srv/AddTwoInts", QosProfile::default())?;
 
     let service_available = r2r::Node::is_available(&client)?;
 

diff --git a/r2r/src/clients.rs b/r2r/src/clients.rs
@@ -6,7 +6,7 @@ use std::{
     sync::{Mutex, Weak},
 };
 
-use crate::{error::*, msg_types::*};
+use crate::{error::*, msg_types::*, QosProfile};
 use r2r_rcl::*;
 
 /// ROS service client.
@@ -319,14 +319,15 @@ impl Client_ for UntypedClient_ {
 }
 
 pub fn create_client_helper(
-    node: *mut rcl_node_t, service_name: &str, service_ts: *const rosidl_service_type_support_t,
+    node: *mut rcl_node_t, service_name: &str, service_ts: *const rosidl_service_type_support_t, qos_profile: QosProfile,
 ) -> Result<rcl_client_t> {
     let mut client_handle = unsafe { rcl_get_zero_initialized_client() };
     let service_name_c_string =
         CString::new(service_name).map_err(|_| Error::RCL_RET_INVALID_ARGUMENT)?;
 
     let result = unsafe {
-        let client_options = rcl_client_get_default_options();
+        let mut client_options = rcl_client_get_default_options();
+        client_options.qos = qos_profile.into();
         rcl_client_init(
             &mut client_handle,
             node,

diff --git a/r2r/src/nodes.rs b/r2r/src/nodes.rs
@@ -277,7 +277,8 @@ impl Node {
             .create_service::<rcl_interfaces::srv::SetParameters::Service>(&format!(
                 "{}/set_parameters",
                 node_name
-            ))?;
+            ),
+            QosProfile::default())?;
 
         let params = self.params.clone();
         let params_struct_clone = params_struct.clone();
@@ -337,10 +338,9 @@ impl Node {
 
         // rcl_interfaces/srv/GetParameters
         let get_params_request_stream = self
-            .create_service::<rcl_interfaces::srv::GetParameters::Service>(&format!(
-                "{}/get_parameters",
-                node_name
-            ))?;
+            .create_service::<rcl_interfaces::srv::GetParameters::Service>(
+                &format!("{}/get_parameters",node_name),
+                QosProfile::default())?;
 
         let params = self.params.clone();
         let params_struct_clone = params_struct.clone();
@@ -379,7 +379,7 @@ impl Node {
         // rcl_interfaces/srv/ListParameters
         use rcl_interfaces::srv::ListParameters;
         let list_params_request_stream = self
-            .create_service::<ListParameters::Service>(&format!("{}/list_parameters", node_name))?;
+            .create_service::<ListParameters::Service>(&format!("{}/list_parameters", node_name), QosProfile::default())?;
 
         let params = self.params.clone();
         let list_params_future = list_params_request_stream.for_each(
@@ -393,8 +393,7 @@ impl Node {
         // rcl_interfaces/srv/DescribeParameters
         use rcl_interfaces::srv::DescribeParameters;
         let desc_params_request_stream = self.create_service::<DescribeParameters::Service>(
-            &format!("{node_name}/describe_parameters"),
-        )?;
+            &format!("{node_name}/describe_parameters"), QosProfile::default())?;
 
         let params = self.params.clone();
         let desc_params_future = desc_params_request_stream.for_each(
@@ -409,7 +408,7 @@ impl Node {
         use rcl_interfaces::srv::GetParameterTypes;
         let get_param_types_request_stream = self.create_service::<GetParameterTypes::Service>(
             &format!("{node_name}/get_parameter_types"),
-        )?;
+        QosProfile::default())?;
 
         let params = self.params.clone();
         let get_param_types_future = get_param_types_request_stream.for_each(
@@ -637,13 +636,13 @@ impl Node {
     /// This function returns a `Stream` of `ServiceRequest`:s. Call
     /// `respond` on the Service Request to send the reply.
     pub fn create_service<T: 'static>(
-        &mut self, service_name: &str,
+        &mut self, service_name: &str, qos_profile: QosProfile,
     ) -> Result<impl Stream<Item = ServiceRequest<T>> + Unpin>
     where
         T: WrappedServiceTypeSupport,
     {
         let service_handle =
-            create_service_helper(self.node_handle.as_mut(), service_name, T::get_ts())?;
+            create_service_helper(self.node_handle.as_mut(), service_name, T::get_ts(), qos_profile)?;
         let (sender, receiver) = mpsc::channel::<ServiceRequest<T>>(10);
 
         let ws = TypedService::<T> {
@@ -659,12 +658,13 @@ impl Node {
     /// Create a ROS service client.
     ///
     /// A service client is used to make requests to a ROS service server.
-    pub fn create_client<T: 'static>(&mut self, service_name: &str) -> Result<Client<T>>
+    pub fn create_client<T: 'static>(
+        &mut self, service_name: &str, qos_profile: QosProfile,) -> Result<Client<T>>
     where
         T: WrappedServiceTypeSupport,
     {
         let client_handle =
-            create_client_helper(self.node_handle.as_mut(), service_name, T::get_ts())?;
+            create_client_helper(self.node_handle.as_mut(), service_name, T::get_ts(), qos_profile)?;
         let ws = TypedClient::<T> {
             rcl_handle: client_handle,
             response_channels: Vec::new(),
@@ -684,11 +684,11 @@ impl Node {
     /// with `serde_json::Value`s instead of concrete types.  Useful
     /// when you cannot know the type of the message at compile time.
     pub fn create_client_untyped(
-        &mut self, service_name: &str, service_type: &str,
+        &mut self, service_name: &str, service_type: &str, qos_profile: QosProfile,
     ) -> Result<ClientUntyped> {
         let service_type = UntypedServiceSupport::new_from(service_type)?;
         let client_handle =
-            create_client_helper(self.node_handle.as_mut(), service_name, service_type.ts)?;
+            create_client_helper(self.node_handle.as_mut(), service_name, service_type.ts, qos_profile)?;
         let client = UntypedClient_ {
             service_type,
             rcl_handle: client_handle,

diff --git a/r2r/src/services.rs b/r2r/src/services.rs
@@ -5,7 +5,7 @@ use std::{
     sync::{Arc, Mutex, Weak},
 };
 
-use crate::{error::*, msg_types::*};
+use crate::{error::*, msg_types::*, QosProfile};
 use r2r_rcl::*;
 
 /// Encapsulates a service request.
@@ -113,14 +113,15 @@ where
 }
 
 pub fn create_service_helper(
-    node: &mut rcl_node_t, service_name: &str, service_ts: *const rosidl_service_type_support_t,
+    node: &mut rcl_node_t, service_name: &str, service_ts: *const rosidl_service_type_support_t, qos_profile: QosProfile,
 ) -> Result<rcl_service_t> {
     let mut service_handle = unsafe { rcl_get_zero_initialized_service() };
     let service_name_c_string =
         CString::new(service_name).map_err(|_| Error::RCL_RET_INVALID_ARGUMENT)?;
 
     let result = unsafe {
-        let service_options = rcl_service_get_default_options();
+        let mut service_options = rcl_service_get_default_options();
+            service_options.qos = qos_profile.into();
         rcl_service_init(
             &mut service_handle,
             node,