Add more documentation for Worker

mxgrey · mxgrey · commit 1d8c46854b5c · 2025-05-14T11:04:58.000Z
Signed-off-by: Michael X. Grey &lt;mxgrey@intrinsic.ai&gt;
diff --git a/examples/minimal_pub_sub/Cargo.toml b/examples/minimal_pub_sub/Cargo.toml
@@ -27,15 +27,9 @@ path = "src/zero_copy_publisher.rs"
 
 [dependencies]
 anyhow = {version = "1", features = ["backtrace"]}
-
-[dependencies.rclrs]
-version = "0.4"
-
-[dependencies.rosidl_runtime_rs]
-version = "0.4"
-
-[dependencies.std_msgs]
-version = "*"
+rclrs = "0.4"
+rosidl_runtime_rs = "0.4"
+example_interfaces = "*"
 
 [package.metadata.ros]
 install_to_share = ["launch"]
diff --git a/examples/minimal_pub_sub/package.xml b/examples/minimal_pub_sub/package.xml
@@ -14,11 +14,11 @@
 
   <build_depend>rclrs</build_depend>
   <build_depend>rosidl_runtime_rs</build_depend>
-  <build_depend>std_msgs</build_depend>
+  <build_depend>example_interfaces</build_depend>
 
   <exec_depend>rclrs</exec_depend>
   <exec_depend>rosidl_runtime_rs</exec_depend>
-  <exec_depend>std_msgs</exec_depend>
+  <exec_depend>example_interfaces</exec_depend>
 
   <export>
     <build_type>ament_cargo</build_type>
diff --git a/examples/minimal_pub_sub/src/minimal_publisher.rs b/examples/minimal_pub_sub/src/minimal_publisher.rs
@@ -7,9 +7,9 @@ fn main() -> Result<(), Error> {
 
     let node = executor.create_node("minimal_publisher")?;
 
-    let publisher = node.create_publisher::<std_msgs::msg::String>("topic")?;
+    let publisher = node.create_publisher::<example_interfaces::msg::String>("topic")?;
 
-    let mut message = std_msgs::msg::String::default();
+    let mut message = example_interfaces::msg::String::default();
 
     let mut publish_count: u32 = 1;
 
diff --git a/examples/minimal_pub_sub/src/minimal_subscriber.rs b/examples/minimal_pub_sub/src/minimal_subscriber.rs
@@ -8,12 +8,11 @@ fn main() -> Result<(), Error> {
     let node = executor.create_node("minimal_subscriber")?;
 
     let worker = node.create_worker::<usize>(0);
-    let _subscription = worker.create_subscription::<std_msgs::msg::String, _>(
+    let _subscription = worker.create_subscription::<example_interfaces::msg::String, _>(
         "topic",
-        move |num_messages: &mut usize, msg: std_msgs::msg::String| {
+        move |num_messages: &mut usize, msg: example_interfaces::msg::String| {
             *num_messages += 1;
-            println!("I heard: '{}'", msg.data);
-            println!("(Got {} messages so far)", *num_messages);
+            println!("#{} | I heard: '{}'", *num_messages, msg.data);
         },
     )?;
 
diff --git a/examples/minimal_pub_sub/src/minimal_two_nodes.rs b/examples/minimal_pub_sub/src/minimal_two_nodes.rs
@@ -5,7 +5,7 @@ use anyhow::{Error, Result};
 
 struct MinimalSubscriberNode {
     #[allow(unused)]
-    subscription: WorkerSubscription<std_msgs::msg::String, SubscriptionData>,
+    subscription: WorkerSubscription<example_interfaces::msg::String, SubscriptionData>,
 }
 
 struct SubscriptionData {
@@ -26,7 +26,7 @@ impl MinimalSubscriberNode {
 
         let subscription = worker.create_subscription(
             topic,
-            |data: &mut SubscriptionData, msg: std_msgs::msg::String| {
+            |data: &mut SubscriptionData, msg: example_interfaces::msg::String| {
                 data.num_messages += 1;
                 println!("[{}] I heard: '{}'", data.node.name(), msg.data);
                 println!(
@@ -50,12 +50,12 @@ fn main() -> Result<(), Error> {
     let _subscriber_node_two =
         MinimalSubscriberNode::new(&executor, "minimal_subscriber_two", "topic")?;
 
-    let publisher = publisher_node.create_publisher::<std_msgs::msg::String>("topic")?;
+    let publisher = publisher_node.create_publisher::<example_interfaces::msg::String>("topic")?;
 
     // TODO(@mxgrey): Replace this with a timer once we have the Timer feature
     // merged in.
     std::thread::spawn(move || -> Result<(), rclrs::RclrsError> {
-        let mut message = std_msgs::msg::String::default();
+        let mut message = example_interfaces::msg::String::default();
         let mut publish_count: u32 = 1;
         loop {
             message.data = format!("Hello, world! {}", publish_count);
diff --git a/examples/minimal_pub_sub/src/zero_copy_publisher.rs b/examples/minimal_pub_sub/src/zero_copy_publisher.rs
@@ -7,7 +7,7 @@ fn main() -> Result<(), Error> {
 
     let node = executor.create_node("minimal_publisher")?;
 
-    let publisher = node.create_publisher::<std_msgs::msg::rmw::UInt32>("topic")?;
+    let publisher = node.create_publisher::<example_interfaces::msg::rmw::UInt32>("topic")?;
 
     let mut publish_count: u32 = 1;
 
diff --git a/examples/minimal_pub_sub/src/zero_copy_subscriber.rs b/examples/minimal_pub_sub/src/zero_copy_subscriber.rs
@@ -7,9 +7,9 @@ fn main() -> Result<(), Error> {
     let node = executor.create_node("minimal_subscriber")?;
 
     let worker = node.create_worker::<usize>(0);
-    let _subscription = worker.create_subscription::<std_msgs::msg::UInt32, _>(
+    let _subscription = worker.create_subscription::<example_interfaces::msg::UInt32, _>(
         "topic",
-        move |num_messages: &mut usize, msg: ReadOnlyLoanedMessage<std_msgs::msg::UInt32>| {
+        move |num_messages: &mut usize, msg: ReadOnlyLoanedMessage<example_interfaces::msg::UInt32>| {
             *num_messages += 1;
             println!("I heard: '{}'", msg.data);
             println!("(Got {} messages so far)", *num_messages);
diff --git a/rclrs/src/client.rs b/rclrs/src/client.rs
@@ -105,14 +105,19 @@ where
     /// when the response is finished being processed, but otherwise you can
     /// safely discard it.
     ///
-    /// # Usage
+    /// # Client Callbacks
     ///
     /// Three callback signatures are supported:
     /// - [`FnOnce`] ( `Response` )
     /// - [`FnOnce`] ( `Response`, [`RequestId`][1] )
     /// - [`FnOnce`] ( `Response`, [`ServiceInfo`] )
     ///
     /// [1]: crate::RequestId
+    ///
+    /// Note that all of these are [`FnOnce`] which grants the greatest amount
+    /// of freedom for what kind of operations you can perform within the
+    /// callback. Just remember that this also means the callbacks are strictly
+    /// one-time-use.
     pub fn call_then<'a, Req, Args>(
         &self,
         request: Req,
@@ -134,7 +139,7 @@ where
     /// when the response is finished being processed, but otherwise you can
     /// safely discard it.
     ///
-    /// # Usage
+    /// # Async Client Callbacks
     ///
     /// Three callback signatures are supported:
     /// - [`FnOnce`] ( `Response` ) -> impl [`Future`][1]<Output=()>
diff --git a/rclrs/src/client/client_async_callback.rs b/rclrs/src/client/client_async_callback.rs
@@ -7,8 +7,11 @@ use crate::{RequestId, ServiceInfo};
 /// A trait to deduce async callbacks of service clients.
 ///
 /// Users of rclrs never need to use this trait directly.
-//
-// TODO(@mxgrey): Add a description of what callback signatures are supported
+///
+/// Three callback signatures are supported:
+/// - [`FnOnce`] ( `Response` ) -> impl [`Future`]<Output=()>
+/// - [`FnOnce`] ( `Response`, [`RequestId`] ) -> impl [`Future`]<Output=()>
+/// - [`FnOnce`] ( `Response`, [`ServiceInfo`] ) -> impl [`Future`]<Output=()>
 pub trait ClientAsyncCallback<T, Args>: Send + 'static
 where
     T: Service,
diff --git a/rclrs/src/client/client_callback.rs b/rclrs/src/client/client_callback.rs
@@ -5,8 +5,11 @@ use crate::{RequestId, ServiceInfo};
 /// A trait to deduce regular callbacks of service clients.
 ///
 /// Users of rclrs never need to use this trait directly.
-//
-// TODO(@mxgrey): Add a description of what callback signatures are supported
+///
+/// Three callback signatures are supported:
+/// - [`FnOnce`] ( `Response` )
+/// - [`FnOnce`] ( `Response`, [`RequestId`] )
+/// - [`FnOnce`] ( `Response`, [`ServiceInfo`] )
 pub trait ClientCallback<T, Args>: Send + 'static
 where
     T: Service,
diff --git a/rclrs/src/executor/basic_executor.rs b/rclrs/src/executor/basic_executor.rs
@@ -17,7 +17,7 @@ use std::{
 
 use crate::{
     WeakActivityListener, ExecutorChannel, ExecutorRuntime, SpinConditions, WorkerChannel,
-    RclrsError, WaitSetRunner, WaitSetRunConditions, Waitable, log_warn, log_fatal,
+    RclrsError, WaitSetRunner, WaitSetRunConditions, Waitable, log_warn, log_debug, log_fatal,
     GuardCondition, ExecutorWorkerOptions, PayloadTask,
 };
 
@@ -37,7 +37,7 @@ static FAILED_TO_SEND_WORKER: &'static str =
 /// a different executor.
 //
 // TODO(@mxgrey): Implement a multi-threaded executor using tokio in a downstream
-// crate and refer to it in this documentation.
+// crate or under a feature gate and refer to it in this documentation.
 pub struct BasicExecutorRuntime {
     ready_queue: Receiver<Arc<Task>>,
     task_sender: TaskSender,
@@ -220,8 +220,6 @@ impl BasicExecutorRuntime {
             let all_guard_conditions = self.all_guard_conditions.clone();
             self.task_sender.add_async_task(Box::pin(async move {
                 if let Err(err) = promise.await {
-                    // TODO(@mxgrey): We should change this to a log when logging
-                    // becomes available.
                     log_warn!(
                         "rclrs.executor.basic_executor",
                         "Sender for SpinOptions::until_promise_resolved was \
@@ -311,16 +309,29 @@ struct BasicWorkerChannel {
 
 impl WorkerChannel for BasicWorkerChannel {
     fn add_to_waitset(&self, new_entity: Waitable) {
-        // TODO(@mxgrey): Log errors here once logging becomes available.
-        self.waitable_sender.unbounded_send(new_entity).ok();
+        if let Err(err) = self.waitable_sender.unbounded_send(new_entity) {
+            // This is a debug log because it is normal for this to happen while
+            // an executor is winding down.
+            log_debug!(
+                "rclrs.basic_executor.add_to_waitset",
+                "Failed to add an item to the waitset: {err}",
+            );
+        }
     }
 
     fn add_async_task(&self, f: BoxFuture<'static, ()>) {
         self.task_sender.add_async_task(f);
     }
 
     fn send_payload_task(&self, f: PayloadTask) {
-        self.payload_task_sender.unbounded_send(f).ok();
+        if let Err(err) = self.payload_task_sender.unbounded_send(f) {
+            // This is a debug log because it is normal for this to happen while
+            // an executor is winding down.
+            log_debug!(
+                "rclrs.BasicWorkerChannel",
+                "Failed to send a payload task: {err}",
+            );
+        }
     }
 
     fn add_activity_listener(&self, listener: WeakActivityListener) {
@@ -340,8 +351,14 @@ impl TaskSender {
             task_sender: self.task_sender.clone(),
         });
 
-        // TODO(@mxgrey): Consider logging errors here once logging is available.
-        self.task_sender.send(task).ok();
+        if let Err(_) = self.task_sender.send(task) {
+            // This is a debug log because it is normal for this to happen while
+            // an executor is winding down.
+            log_debug!(
+                "rclrs.TaskSender.add_async_task",
+                "Failed to send a task. This indicates the Worker has shut down.",
+            );
+        }
     }
 }
 
diff --git a/rclrs/src/node.rs b/rclrs/src/node.rs
@@ -587,7 +587,7 @@ impl NodeState {
     /// ```
     /// # use rclrs::*;
     /// # let executor = Context::default().create_basic_executor();
-    /// # let node = executor.create_node("my_node").unwrap();
+    /// # let node = executor.create_node("my_node")?;
     /// use std::sync::Arc;
     /// use example_interfaces::srv::*;
     ///
@@ -618,7 +618,8 @@ impl NodeState {
     ///             AddTwoInts_Response { sum: a + b }
     ///        }
     ///     }
-    /// );
+    /// )?;
+    /// # Ok::<(), RclrsError>(())
     /// ```
     pub fn create_async_service<'a, T, Args>(
         &self,
@@ -760,6 +761,12 @@ impl NodeState {
     /// });
     /// # Ok::<(), RclrsError>(())
     /// ```
+    ///
+    /// You can change the subscription at any time by calling
+    /// [`SubscriptionState::set_callback`] or
+    /// [`SubscriptionState::set_async_callback`]. Even if the subscription is
+    /// initially created with a regular callback, it can be changed to an async
+    /// callback at any time.
     pub fn create_subscription<'a, T, Args>(
         &self,
         options: impl Into<SubscriptionOptions<'a>>,
@@ -862,6 +869,12 @@ impl NodeState {
     /// )?;
     /// # Ok::<(), RclrsError>(())
     /// ```
+    ///
+    /// You can change the subscription at any time by calling
+    /// [`SubscriptionState::set_callback`] or
+    /// [`SubscriptionState::set_async_callback`]. Even if the subscription is
+    /// initially created with an async callback, it can be changed to a regular
+    /// callback at any time.
     pub fn create_async_subscription<'a, T, Args>(
         &self,
         options: impl Into<SubscriptionOptions<'a>>,
diff --git a/rclrs/src/service/into_async_service_callback.rs b/rclrs/src/service/into_async_service_callback.rs
@@ -6,7 +6,10 @@ use std::future::Future;
 
 /// A trait for async callbacks of services.
 ///
-// TODO(@mxgrey): Add a description of what callbacks signatures are supported
+/// Three callback signatures are supported:
+/// - [`FnMut`] ( `Request` ) -> impl [`Future`]<Output=`Response`>
+/// - [`FnMut`] ( `Request`, [`RequestId`] ) -> impl [`Future`]<Output=`Response`>
+/// - [`FnMut`] ( `Request`, [`ServiceInfo`] ) -> impl [`Future`]<Output=`Response`>
 pub trait IntoAsyncServiceCallback<T, Args>: Send + 'static
 where
     T: Service,
diff --git a/rclrs/src/service/into_node_service_callback.rs b/rclrs/src/service/into_node_service_callback.rs
@@ -7,8 +7,11 @@ use std::sync::Arc;
 /// A trait to deduce regular callbacks of services.
 ///
 /// Users of rclrs never need to use this trait directly.
-//
-// TODO(@mxgrey): Add a description of what callbacks signatures are supported
+///
+/// Three callback signatures are supported:
+/// - [`Fn`] ( `Request` ) -> `Response`
+/// - [`Fn`] ( `Request`, [`RequestId`] ) -> `Response`
+/// - [`Fn`] ( `Request`, [`ServiceInfo`] ) -> `Response`
 pub trait IntoNodeServiceCallback<T, Args>: Send + 'static
 where
     T: Service,
diff --git a/rclrs/src/service/into_worker_service_callback.rs b/rclrs/src/service/into_worker_service_callback.rs
@@ -8,8 +8,14 @@ use crate::{
 /// A trait used to deduce callbacks for services that run on a worker.
 ///
 /// Users of rclrs never need to use this trait directly.
-//
-// TODO(@mxgrey): Add a description of what callback signatures are supported
+///
+/// Worker service callbacks support six signatures:
+/// - [`FnMut`] ( `Request` ) -> `Response`
+/// - [`FnMut`] ( `Request`, [`RequestId`] ) -> `Response`
+/// - [`FnMut`] ( `Request`, [`ServiceInfo`] ) -> `Response`
+/// - [`FnMut`] ( `&mut Payload`, `Request` ) -> `Response`
+/// - [`FnMut`] ( `&mut Payload`, `Request`,  [`RequestId`] ) -> `Response`
+/// - [`FnMut`] ( `&mut Payload`, `Request`, [`ServiceInfo`] ) -> `Response`
 pub trait IntoWorkerServiceCallback<T, Payload, Args>: Send + 'static
 where
     T: Service,
diff --git a/rclrs/src/worker.rs b/rclrs/src/worker.rs
