Worker subscriptions

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

Author: luca-della-vedova

rclrs/src/dynamic_message/dynamic_subscription.rs

@@ -14,11 +14,13 @@ use crate::rcl_bindings::*;
 use crate::{
     MessageInfo, Node, NodeHandle, RclPrimitive, RclPrimitiveHandle, RclPrimitiveKind, RclrsError,
     SubscriptionHandle, SubscriptionOptions, ToResult, Waitable, WaitableLifecycle, WorkScope,
-    WorkerCommands, ENTITY_LIFECYCLE_MUTEX,
+    Worker, WorkerCommands, ENTITY_LIFECYCLE_MUTEX,
 };
 
 pub type DynamicSubscription = Arc<DynamicSubscriptionState<Node>>;
 
+pub type WorkerDynamicSubscription<Payload> = Arc<DynamicSubscriptionState<Worker<Payload>>>;
+
 struct DynamicSubscriptionExecutable<Payload> {
     handle: Arc<SubscriptionHandle>,
     callback: Arc<Mutex<DynamicSubscriptionCallback<Payload>>>,

rclrs/src/node.rs

@@ -812,14 +812,13 @@ impl NodeState {
     where
         F: FnMut(DynamicMessage, MessageInfo) -> BoxFuture<'static, ()> + Send + 'static,
     {
-        let subscription = DynamicSubscriptionState::<Node>::create(
+        DynamicSubscriptionState::<Node>::create(
             topic_type,
             options,
             NodeDynamicSubscriptionCallback(Box::new(callback)),
             &self.handle,
             self.commands.async_worker_commands(),
-        )?;
-        Ok(subscription)
+        )
     }
 
     /// Creates a [`Subscription`] with an async callback.

rclrs/src/worker.rs

@@ -1,6 +1,11 @@
+#[cfg(feature = "dyn_msg")]
+use crate::dynamic_message::{
+    DynamicMessage, DynamicSubscriptionState, MessageTypeName, WorkerDynamicSubscription,
+    WorkerDynamicSubscriptionCallback,
+};
 use crate::{
-    log_fatal, IntoWorkerServiceCallback, IntoWorkerSubscriptionCallback, Node, Promise,
-    RclrsError, ServiceOptions, ServiceState, SubscriptionOptions, SubscriptionState,
+    log_fatal, IntoWorkerServiceCallback, IntoWorkerSubscriptionCallback, MessageInfo, Node,
+    Promise, RclrsError, ServiceOptions, ServiceState, SubscriptionOptions, SubscriptionState,
     WorkerCommands, WorkerService, WorkerSubscription,
 };
 use futures::channel::oneshot;
@@ -264,6 +269,25 @@ impl<Payload: 'static + Send + Sync> WorkerState<Payload> {
         )
     }
 
+    #[cfg(feature = "dyn_msg")]
+    pub fn create_dynamic_subscription<'a, F>(
+        &self,
+        topic_type: MessageTypeName,
+        options: impl Into<SubscriptionOptions<'a>>,
+        callback: F,
+    ) -> Result<WorkerDynamicSubscription<Payload>, RclrsError>
+    where
+        F: FnMut(&mut Payload, DynamicMessage, MessageInfo) + Send + 'static,
+    {
+        DynamicSubscriptionState::<Worker<Payload>>::create(
+            topic_type,
+            options,
+            WorkerDynamicSubscriptionCallback(Box::new(callback)),
+            self.node.handle(),
+            &self.commands,
+        )
+    }
+
     /// Creates a [`WorkerService`].
     ///
     /// Unlike services created from a [`Node`], the callbacks for these services