Add docs, cleanup tests

Signed-off-by: Luca Della Vedova <[email protected]>

Author: luca-della-vedova

rclrs/src/dynamic_message/dynamic_publisher.rs

@@ -9,6 +9,12 @@ use crate::error::{RclrsError, ToResult};
 use crate::rcl_bindings::*;
 use crate::{NodeHandle, PublisherHandle, PublisherOptions, ENTITY_LIFECYCLE_MUTEX};
 
+/// Struct for sending dynamic messages.
+///
+/// Create a dynamic publisher using [`Node::create_dynamic_publisher`][1].
+/// Refer to [`crate::Publisher`] for details of the behavior.
+///
+/// [1]: crate::NodeState::create_dynamic_publisher
 pub type DynamicPublisher = Arc<DynamicPublisherState>;
 
 /// Struct for sending messages of type `T`.
@@ -158,17 +164,17 @@ mod tests {
         let graph = construct_test_graph(namespace)?;
 
         let node_1_empty_publisher = graph.node1.create_dynamic_publisher(
-            "test_msgs/msg/Empty".try_into().unwrap(),
+            "test_msgs/msg/Empty".try_into()?,
             "graph_test_topic_1",
         )?;
         let topic1 = node_1_empty_publisher.topic_name();
         let node_1_basic_types_publisher = graph.node1.create_dynamic_publisher(
-            "test_msgs/msg/BasicTypes".try_into().unwrap(),
+            "test_msgs/msg/BasicTypes".try_into()?,
             "graph_test_topic_2",
         )?;
         let topic2 = node_1_basic_types_publisher.topic_name();
         let node_2_default_publisher = graph.node2.create_dynamic_publisher(
-            "test_msgs/msg/Defaults".try_into().unwrap(),
+            "test_msgs/msg/Defaults".try_into()?,
             "graph_test_topic_3",
         )?;
         let topic3 = node_2_default_publisher.topic_name();

rclrs/src/dynamic_message/dynamic_subscription.rs

@@ -17,8 +17,30 @@ use crate::{
     Worker, WorkerCommands, ENTITY_LIFECYCLE_MUTEX,
 };
 
+/// Struct for receiving messages whose type is not know at compile time.
+///
+/// Create a dynamic subscription using [`NodeState::create_dynamic_subscription()`][1]
+/// or [`NodeState::create_async_dynamic_subscription`][2].
+///
+/// There can be multiple subscriptions for the same topic, in different nodes or the same node.
+/// A clone of a `Subscription` will refer to the same subscription instance as the original.
+/// The underlying instance is tied to [`DynamicSubscriptionState`] which implements the [`DynamicSubscription`] API.
+///
+/// Receiving messages requires the node's executor to [spin][3].
+///
+/// When a subscription is created, it may take some time to get "matched" with a corresponding
+/// publisher.
+///
+/// [1]: crate::NodeState::create_subscription
+/// [2]: crate::NodeState::create_async_subscription
+/// [3]: crate::Executor::spin
 pub type DynamicSubscription = Arc<DynamicSubscriptionState<Node>>;
 
+/// A [`DynamicSubscription`] that runs on a [`Worker`].
+///
+/// Create a worker dynamic subscription using [`WorkerState::create_dynamic_subscription`][1].
+///
+/// [1]: crate::WorkerState::create_dynamic_subscription
 pub type WorkerDynamicSubscription<Payload> = Arc<DynamicSubscriptionState<Worker<Payload>>>;
 
 struct DynamicSubscriptionExecutable<Payload> {
@@ -165,7 +187,7 @@ impl<Payload: 'static> DynamicSubscriptionCallback<Payload> {
 }
 
 impl<Payload> DynamicSubscriptionExecutable<Payload> {
-    pub fn take(&self) -> Result<(DynamicMessage, MessageInfo), RclrsError> {
+    fn take(&self) -> Result<(DynamicMessage, MessageInfo), RclrsError> {
         let mut dynamic_message = self.metadata.create()?;
         let rmw_message = dynamic_message.storage.as_mut_ptr();
         let mut message_info = unsafe { rmw_get_zero_initialized_message_info() };
@@ -357,20 +379,20 @@ mod tests {
         let graph = construct_test_graph(namespace)?;
 
         let node_2_empty_subscription = graph.node2.create_dynamic_subscription::<_>(
-            "test_msgs/msg/Empty".try_into().unwrap(),
+            "test_msgs/msg/Empty".try_into()?,
             "graph_test_topic_1",
             |_, _| {},
         )?;
         let topic1 = node_2_empty_subscription.topic_name();
         let node_2_basic_types_subscription = graph.node2.create_dynamic_subscription::<_>(
-            "test_msgs/msg/BasicTypes".try_into().unwrap(),
+            "test_msgs/msg/BasicTypes".try_into()?,
             "graph_test_topic_2",
             |_, _| {},
         )?;
         let topic2 = node_2_basic_types_subscription.topic_name();
 
         let node_1_defaults_subscription = graph.node1.create_dynamic_subscription::<_>(
-            "test_msgs/msg/Defaults".try_into().unwrap(),
+            "test_msgs/msg/Defaults".try_into()?,
             "graph_test_topic_3",
             |_, _| {},
         )?;

rclrs/src/lib.rs

@@ -200,6 +200,8 @@ mod rcl_bindings;
 
 #[cfg(feature = "dyn_msg")]
 pub mod dynamic_message;
+#[cfg(feature = "dyn_msg")]
+pub use dynamic_message::*;
 
 pub use arguments::*;
 pub use client::*;

rclrs/src/node.rs

@@ -403,6 +403,22 @@ impl NodeState {
         PublisherState::<T>::create(options, Arc::clone(&self.handle))
     }
 
+    /// Creates a [`DynamicPublisher`], a publisher whose type is only known at runtime.
+    ///
+    /// Refer to [`Node::create_publisher`] for the API, the only key difference is that the
+    /// publisher's message type is passed as a [`crate::MessageTypeName`] parameter.
+    ///
+    /// Pass in only the topic name for the `options` argument to use all default publisher options:
+    /// ```
+    /// # use rclrs::*;
+    /// # let executor = Context::default().create_basic_executor();
+    /// # let node = executor.create_node("my_node").unwrap();
+    /// let publisher = node.create_dynamic_publisher(
+    ///     "test_msgs/msg/Empty".try_into().unwrap(),
+    ///     "my_topic"
+    ///     .keep_last(100)
+    /// )
+    /// .unwrap();
     #[cfg(feature = "dyn_msg")]
     pub fn create_dynamic_publisher<'a>(
         &self,
@@ -803,6 +819,42 @@ impl NodeState {
         )
     }
 
+    /// Creates a [`DynamicSubscription`] with an ordinary callback.
+    ///
+    /// For the behavior and API refer to [`Node::create_subscription`], except two key
+    /// differences:
+    ///
+    ///   - The message type is determined at runtime through the `topic_type` function parameter.
+    ///   - Only one type of callback is supported (returning both [`crate::DynamicMessage`] and
+    ///   [`crate::MessageInfo`]).
+    ///
+    /// # Message type passing
+    ///
+    /// The message type can be passed as a [`crate::MessageTypeName`] struct. The struct also implements `TryFrom<&str>`
+    /// ```
+    /// # use rclrs::*;
+    /// # let executor = Context::default().create_basic_executor();
+    /// # let node = executor.create_node("my_node").unwrap();
+    /// let subscription = node.create_dynamic_subscription(
+    ///     MessageTypeName {
+    ///         package_name: "test_msgs".to_owned(),
+    ///         type_name: "Empty".to_owned(),
+    ///     },
+    ///     "my_topic"
+    ///     .transient_local(),
+    ///     |_msg: DynamicMessage, _info: MessageInfo| {
+    ///         println!("Received message!");
+    ///     },
+    /// );
+    ///
+    /// let subscription = node.create_dynamic_subscription(
+    ///     "test_msgs/msg/Empty".try_into().unwrap(),
+    ///     "my_topic",
+    ///     |_msg: DynamicMessage, _info: MessageInfo| {
+    ///         println!("Received message!");
+    ///     },
+    /// );
+    /// ```
     #[cfg(feature = "dyn_msg")]
     pub fn create_dynamic_subscription<'a, F>(
         &self,
@@ -822,6 +874,60 @@ impl NodeState {
         )
     }
 
+    /// Creates a [`DynamicSubscription`] with an async callback.
+    ///
+    /// For the behavior and API refer to [`Node::create_async_subscription`], except two key
+    /// differences:
+    ///
+    ///   - The message type is determined at runtime through the `topic_type` function parameter.
+    ///   - Only one type of callback is supported (returning both [`crate::DynamicMessage`] and
+    ///   [`crate::MessageInfo`].
+    ///
+    /// # Message type passing
+    ///
+    /// The message type can be passed as a [`crate::MessageTypeName`] struct. The struct also implements `TryFrom<&str>`
+    /// ```
+    /// # use rclrs::*;
+    /// # let executor = Context::default().create_basic_executor();
+    /// # let node = executor.create_node("my_node").unwrap();
+    /// use std::sync::Arc;
+    ///
+    /// let count_worker = node.create_worker(0_usize);
+    /// let data_worker = node.create_worker(String::new());
+    ///
+    /// let service = node.create_async_dynamic_subscription(
+    ///     "example_interfaces/msg/String".try_into()?,
+    ///     "topic",
+    ///     move |msg: DynamicMessage, _info: MessageInfo| {
+    ///         // Clone the workers so they can be captured into the async block
+    ///         let count_worker = Arc::clone(&count_worker);
+    ///         let data_worker = Arc::clone(&data_worker);
+    ///         Box::pin(async move {
+    ///             // Update the message count
+    ///             let current_count = count_worker.run(move |count: &mut usize| {
+    ///                 *count += 1;
+    ///                 *count
+    ///             }).await.unwrap();
+    ///
+    ///             // Change the data in the data_worker and get back the data
+    ///             // that was previously put in there.
+    ///             let previous = data_worker.run(move |data: &mut String| {
+    ///                 let value = msg.get("data").unwrap();
+    ///                 let Value::Simple(value) = value else {
+    ///                     panic!("Unexpected value type, expected Simple value");
+    ///                 };
+    ///                 let SimpleValue::String(value) = value else {
+    ///                     panic!("Unexpected value type, expected String");
+    ///                 };
+    ///                 std::mem::replace(data, value.to_string())
+    ///             }).await.unwrap();
+    ///
+    ///             println!("Current count is {current_count}, data was previously {previous}");
+    ///        })
+    ///     }
+    /// )?;
+    /// # Ok::<(), RclrsError>(())
+    /// ```
     #[cfg(feature = "dyn_msg")]
     pub fn create_async_dynamic_subscription<'a, F>(
         &self,

rclrs/src/worker.rs

@@ -269,6 +269,45 @@ impl<Payload: 'static + Send + Sync> WorkerState<Payload> {
         )
     }
 
+    /// Creates a [`WorkerDynamicSubscription`], whose message type is only know at runtime.
+    ///
+    /// Refer to ['Worker::create_subscription`] for the API and behavior except two key
+    /// differences:
+    ///
+    ///   - The message type is determined at runtime through the `topic_type` function parameter.
+    ///   - Only one type of callback is supported (returning both [`crate::DynamicMessage`] and
+    ///   [`crate::MessageInfo`]).
+    ///
+    /// ```
+    /// # use rclrs::*;
+    /// # let executor = Context::default().create_basic_executor();
+    /// # let node = executor.create_node("my_node").unwrap();
+    /// // The worker's payload is data that we want to share with other callbacks.
+    /// let worker = node.create_worker::<Option<String>>(None);
+    ///
+    /// // This variable will be the mutable internal state of the subscription
+    /// // callback.
+    /// let mut count = 0_usize;
+    ///
+    /// let subscription = worker.create_dynamic_subscription(
+    ///     "example_interfaces/msg/String".try_into()?,
+    ///     "topic",
+    ///     move |data: &mut Option<String>, msg: DynamicMessage, _msg_info: MessageInfo| {
+    ///         count += 1;
+    ///         let value = msg.get("data").unwrap();
+    ///         let Value::Simple(value) = value else {
+    ///             panic!("Unexpected value type, expected Simple value");
+    ///         };
+    ///         let SimpleValue::String(value) = value else {
+    ///             panic!("Unexpected value type, expected String");
+    ///         };
+    ///         println!("#{count} | I heard: '{}'", value);
+    ///
+    ///         *data = Some(value.to_string());
+    ///     },
+    /// )?;
+    /// # Ok::<(), RclrsError>(())
+    /// ```
     #[cfg(feature = "dyn_msg")]
     pub fn create_dynamic_subscription<'a, F>(
         &self,
@@ -535,6 +574,7 @@ mod tests {
     struct TestPayload {
         subscription_count: usize,
         service_count: usize,
+        #[cfg(feature = "dyn_msg")]
         dynamic_subscription_count: usize,
     }
 
@@ -550,6 +590,7 @@ mod tests {
             },
         );
 
+        #[cfg(feature = "dyn_msg")]
         let _count_dynamic_sub = worker.create_dynamic_subscription(
             "test_msgs/msg/Empty".try_into().unwrap(),
             "test_worker_topic",
@@ -569,8 +610,14 @@ mod tests {
         let promise = worker.listen_until(move |payload| {
             if payload.service_count > 0
                 && payload.subscription_count > 0
-                && payload.dynamic_subscription_count > 0
             {
+                #[cfg(feature = "dyn_msg")]
+                if payload.dynamic_subscription_count > 0 {
+                    Some(*payload)
+                } else {
+                    None
+                }
+                #[cfg(not(feature = "dyn_msg"))]
                 Some(*payload)
             } else {
                 None