API symmetry with normal pub/sub

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

Author: luca-della-vedova

rclrs/src/dynamic_message/dynamic_publisher.rs

@@ -9,6 +9,8 @@ use crate::error::{RclrsError, ToResult};
 use crate::rcl_bindings::*;
 use crate::{NodeHandle, PublisherHandle, PublisherOptions, ENTITY_LIFECYCLE_MUTEX};
 
+pub type DynamicPublisher = Arc<DynamicPublisherState>;
+
 /// Struct for sending messages of type `T`.
 ///
 /// Multiple publishers can be created for the same topic, in different nodes or the same node.
@@ -19,23 +21,23 @@ use crate::{NodeHandle, PublisherHandle, PublisherOptions, ENTITY_LIFECYCLE_MUTE
 /// Sending messages does not require calling [`spin`][1] on the publisher's node.
 ///
 /// [1]: crate::spin
-pub struct DynamicPublisher {
+pub struct DynamicPublisherState {
     handle: PublisherHandle,
     metadata: DynamicMessageMetadata,
     // This is the regular type support library, not the introspection one.
     #[allow(dead_code)]
     type_support_library: Arc<libloading::Library>,
 }
 
-impl DynamicPublisher {
-    /// Creates a new `DynamicPublisher`.
+impl DynamicPublisherState {
+    /// Creates a new `DynamicPublisherState`.
     ///
     /// Node and namespace changes are always applied _before_ topic remapping.
-    pub(crate) fn new<'a>(
+    pub(crate) fn create<'a>(
         topic_type: MessageTypeName,
         options: impl Into<PublisherOptions<'a>>,
         node_handle: Arc<NodeHandle>,
-    ) -> Result<Self, RclrsError> {
+    ) -> Result<Arc<Self>, RclrsError> {
         // This loads the introspection type support library.
         let metadata = DynamicMessageMetadata::new(topic_type)?;
         let PublisherOptions { topic, qos } = options.into();
@@ -86,14 +88,14 @@ impl DynamicPublisher {
             }
         }
 
-        Ok(Self {
+        Ok(Arc::new(Self {
             handle: PublisherHandle {
                 rcl_publisher: Mutex::new(rcl_publisher),
                 node_handle,
             },
             metadata,
             type_support_library,
-        })
+        }))
     }
 
     /// Returns the topic name of the publisher.

rclrs/src/dynamic_message/dynamic_subscription.rs

@@ -12,11 +12,13 @@ use super::{
 };
 use crate::rcl_bindings::*;
 use crate::{
-    MessageInfo, NodeHandle, RclPrimitive, RclPrimitiveHandle, RclPrimitiveKind, RclrsError,
+    MessageInfo, Node, NodeHandle, RclPrimitive, RclPrimitiveHandle, RclPrimitiveKind, RclrsError,
     SubscriptionHandle, SubscriptionOptions, ToResult, Waitable, WaitableLifecycle, WorkScope,
     WorkerCommands, ENTITY_LIFECYCLE_MUTEX,
 };
 
+pub type DynamicSubscription = Arc<DynamicSubscriptionState<Node>>;
+
 struct DynamicSubscriptionExecutable<Payload> {
     handle: Arc<SubscriptionHandle>,
     callback: Arc<Mutex<DynamicSubscriptionCallback<Payload>>>,
@@ -153,7 +155,7 @@ impl<Payload: 'static> RclPrimitive for DynamicSubscriptionExecutable<Payload> {
 }
 
 /// Struct for receiving messages whose type is only known at runtime.
-pub struct DynamicSubscription<Scope>
+pub struct DynamicSubscriptionState<Scope>
 where
     Scope: WorkScope,
 {
@@ -174,14 +176,14 @@ where
     type_support_library: Arc<libloading::Library>,
 }
 
-impl<Scope> DynamicSubscription<Scope>
+impl<Scope> DynamicSubscriptionState<Scope>
 where
     Scope: WorkScope,
 {
     /// Creates a new dynamic subscription.
     ///
     /// This is not a public function, by the same rationale as `Subscription::new()`.
-    pub(crate) fn new<'a>(
+    pub(crate) fn create<'a>(
         topic_type: MessageTypeName,
         options: impl Into<SubscriptionOptions<'a>>,
         callback: impl Into<DynamicSubscriptionCallback<Scope::Payload>>,
@@ -336,11 +338,9 @@ mod tests {
     fn assert_send<T: Send>() {}
     fn assert_sync<T: Sync>() {}
 
-    /*
     #[test]
     fn dynamic_subscription_is_sync_and_send() {
         assert_send::<DynamicSubscription>();
         assert_sync::<DynamicSubscription>();
     }
-    */
 }

rclrs/src/node.rs

@@ -6,7 +6,10 @@ pub use primitive_options::*;
 
 mod graph;
 #[cfg(feature = "dyn_msg")]
-use crate::dynamic_message::{DynamicMessage, DynamicPublisher, DynamicSubscription};
+use crate::dynamic_message::{
+    DynamicMessage, DynamicPublisher, DynamicPublisherState, DynamicSubscription,
+    DynamicSubscriptionState, MessageTypeName, NodeDynamicSubscriptionCallback,
+};
 
 pub use graph::*;
 
@@ -33,9 +36,7 @@ use async_std::future::timeout;
 use rosidl_runtime_rs::Message;
 
 use crate::{
-    dynamic_message::{MessageTypeName, NodeDynamicSubscriptionCallback},
-    rcl_bindings::*,
-    Client, ClientOptions, ClientState, Clock, ContextHandle, ExecutorCommands,
+    rcl_bindings::*, Client, ClientOptions, ClientState, Clock, ContextHandle, ExecutorCommands,
     IntoAsyncServiceCallback, IntoAsyncSubscriptionCallback, IntoNodeServiceCallback,
     IntoNodeSubscriptionCallback, LogParams, Logger, MessageInfo, ParameterBuilder,
     ParameterInterface, ParameterVariant, Parameters, Promise, Publisher, PublisherOptions,
@@ -407,7 +408,7 @@ impl NodeState {
         topic_type: MessageTypeName,
         options: impl Into<PublisherOptions<'a>>,
     ) -> Result<DynamicPublisher, RclrsError> {
-        DynamicPublisher::new(topic_type, options, Arc::clone(&self.handle))
+        DynamicPublisherState::create(topic_type, options, Arc::clone(&self.handle))
     }
 
     /// Creates a [`Service`] with an ordinary callback.
@@ -807,18 +808,17 @@ impl NodeState {
         topic_type: MessageTypeName,
         options: impl Into<SubscriptionOptions<'a>>,
         callback: F,
-    ) -> Result<Arc<DynamicSubscription<Node>>, RclrsError>
+    ) -> Result<DynamicSubscription, RclrsError>
     where
         F: FnMut(DynamicMessage, MessageInfo) -> BoxFuture<'static, ()> + Send + 'static,
     {
-        let subscription = DynamicSubscription::new(
+        let subscription = DynamicSubscriptionState::<Node>::create(
             topic_type,
             options,
             NodeDynamicSubscriptionCallback(Box::new(callback)),
             &self.handle,
             self.commands.async_worker_commands(),
         )?;
-        // TODO(luca)  similar API to above?
         Ok(subscription)
     }