Migrated all primitives to new worker architecture

Signed-off-by: Michael X. Grey <[email protected]>

Author: mxgrey

rclrs/src/client.rs

@@ -185,8 +185,8 @@ where
 
     /// Creates a new client.
     pub(crate) fn create<'a>(
-        node: &Node,
         options: impl Into<ClientOptions<'a>>,
+        node: &Node,
     ) -> Result<Arc<Self>, RclrsError>
     // This uses pub(crate) visibility to avoid instantiating this struct outside
     // [`Node::create_client`], see the struct's documentation for the rationale
@@ -231,12 +231,12 @@ where
             }
         }
 
+        let commands = node.commands().async_worker_commands();
         let handle = Arc::new(ClientHandle {
             rcl_client: Mutex::new(rcl_client),
             node: Arc::clone(&node),
         });
 
-        let commands = node.commands();
         let board = Arc::new(Mutex::new(ClientRequestBoard::new()));
 
         let (waitable, lifecycle) = Waitable::new(
@@ -300,7 +300,7 @@ impl<T> RclPrimitive for ClientExecutable<T>
 where
     T: rosidl_runtime_rs::Service,
 {
-    unsafe fn execute(&mut self, payload: &mut dyn Any) -> Result<(), RclrsError> {
+    unsafe fn execute(&mut self, _: &mut dyn Any) -> Result<(), RclrsError> {
         self.board.lock().unwrap().execute(&self.handle)
     }
 

rclrs/src/executor.rs

@@ -91,7 +91,7 @@ impl Executor {
         E: 'static + ExecutorRuntime + Send,
     {
         let executor_channel = runtime.channel();
-        let node_worker_commands = ExecutorCommands::impl_create_worker_commands(
+        let async_worker_commands = ExecutorCommands::impl_create_worker_commands(
             &Context { handle: Arc::clone(&context) },
             &*executor_channel,
             Box::new(()),
@@ -103,7 +103,7 @@ impl Executor {
             },
             executor_channel,
             halt_spinning: Arc::new(AtomicBool::new(false)),
-            node_worker_commands,
+            async_worker_commands,
         });
 
         Self {
@@ -130,7 +130,7 @@ impl Executor {
 pub struct ExecutorCommands {
     context: Context,
     executor_channel: Arc<dyn ExecutorChannel>,
-    node_worker_commands: Arc<WorkerCommands>,
+    async_worker_commands: Arc<WorkerCommands>,
     halt_spinning: Arc<AtomicBool>,
 }
 
@@ -169,7 +169,7 @@ impl ExecutorCommands {
         F::Output: Send,
     {
         let (mut sender, receiver) = oneshot::channel();
-        self.node_worker_commands.channel.add_async_task(Box::pin(async move {
+        self.async_worker_commands.channel.add_async_task(Box::pin(async move {
             let cancellation = sender.cancellation();
             let output = match select(cancellation, std::pin::pin!(f)).await {
                 // The task was cancelled
@@ -204,7 +204,7 @@ impl ExecutorCommands {
         F::Output: Send,
     {
         let (sender, receiver) = oneshot::channel();
-        self.node_worker_commands.channel.add_async_task(Box::pin(async move {
+        self.async_worker_commands.channel.add_async_task(Box::pin(async move {
             sender.send(f.await).ok();
         }));
         receiver
@@ -216,23 +216,23 @@ impl ExecutorCommands {
     }
 
     pub(crate) fn add_to_wait_set(&self, waitable: Waitable) {
-        self.node_worker_commands.channel.add_to_waitset(waitable);
+        self.async_worker_commands.channel.add_to_waitset(waitable);
     }
 
     #[cfg(test)]
     pub(crate) fn wake_all_wait_sets(&self) {
         self.executor_channel.wake_all_wait_sets();
     }
 
-    pub(crate) fn node_worker_commands(&self) -> &Arc<WorkerCommands> {
-        &self.node_worker_commands
+    pub(crate) fn async_worker_commands(&self) -> &Arc<WorkerCommands> {
+        &self.async_worker_commands
     }
 
     pub(crate) fn create_worker_commands(&self, payload: Box<dyn Any + Send>) -> Arc<WorkerCommands> {
         Self::impl_create_worker_commands(&self.context, &*self.executor_channel, payload)
     }
 
-    /// We separate out this impl function so that we can create the node worker
+    /// We separate out this impl function so that we can create the async worker
     /// before the [`ExecutorCommands`] is finished being constructed.
     fn impl_create_worker_commands(
         context: &Context,

rclrs/src/node.rs

@@ -31,8 +31,8 @@ use rosidl_runtime_rs::Message;
 use crate::{
     rcl_bindings::*, Client, ClientOptions, ClientState, Clock, ContextHandle, ExecutorCommands,
     LogParams, Logger, ParameterBuilder, ParameterInterface, ParameterVariant, Parameters, Promise,
-    Publisher, PublisherOptions, PublisherState, RclrsError, Service, ServiceAsyncCallback,
-    ServiceCallback, ServiceOptions, ServiceState, Subscription, IntoAsyncSubscriptionCallback,
+    Publisher, PublisherOptions, PublisherState, RclrsError, Service, IntoAsyncServiceCallback,
+    IntoNodeServiceCallback, ServiceOptions, ServiceState, Subscription, IntoAsyncSubscriptionCallback,
     IntoNodeSubscriptionCallback, SubscriptionOptions, SubscriptionState, TimeSource, ToLogParams,
     ENTITY_LIFECYCLE_MUTEX,
 };
@@ -278,7 +278,7 @@ impl NodeState {
     where
         T: rosidl_runtime_rs::Service,
     {
-        ClientState::<T>::create(self, options)
+        ClientState::<T>::create(options, self)
     }
 
     /// Creates a [`Publisher`][1].
@@ -322,7 +322,7 @@ impl NodeState {
     where
         T: Message,
     {
-        PublisherState::<T>::create(Arc::clone(&self.handle), options)
+        PublisherState::<T>::create(options, Arc::clone(&self.handle))
     }
 
     /// Creates a [`Service`] with an ordinary callback.
@@ -385,16 +385,16 @@ impl NodeState {
     pub fn create_service<'a, T, Args>(
         &self,
         options: impl Into<ServiceOptions<'a>>,
-        callback: impl ServiceCallback<T, Args>,
+        callback: impl IntoNodeServiceCallback<T, Args>,
     ) -> Result<Service<T>, RclrsError>
     where
         T: rosidl_runtime_rs::Service,
     {
-        ServiceState::<T>::create(
+        ServiceState::<T, Node>::create(
             options,
-            callback.into_service_callback(),
+            callback.into_node_service_callback(),
             &self.handle,
-            &self.commands,
+            self.commands.async_worker_commands(),
         )
     }
 
@@ -420,16 +420,16 @@ impl NodeState {
     pub fn create_async_service<'a, T, Args>(
         &self,
         options: impl Into<ServiceOptions<'a>>,
-        callback: impl ServiceAsyncCallback<T, Args>,
+        callback: impl IntoAsyncServiceCallback<T, Args>,
     ) -> Result<Service<T>, RclrsError>
     where
         T: rosidl_runtime_rs::Service,
     {
-        ServiceState::<T>::create(
+        ServiceState::<T, Node>::create(
             options,
-            callback.into_service_async_callback(),
+            callback.into_async_service_callback(),
             &self.handle,
-            &self.commands,
+            self.commands.async_worker_commands(),
         )
     }
 
@@ -501,7 +501,7 @@ impl NodeState {
             options,
             callback.into_node_subscription_callback(),
             &self.handle,
-            self.commands.node_worker_commands(),
+            self.commands.async_worker_commands(),
         )
     }
 
@@ -536,7 +536,7 @@ impl NodeState {
             options,
             callback.into_async_subscription_callback(),
             &self.handle,
-            self.commands.node_worker_commands(),
+            self.commands.async_worker_commands(),
         )
     }
 

rclrs/src/publisher.rs

@@ -96,8 +96,8 @@ where
     ///
     /// Node and namespace changes are always applied _before_ topic remapping.
     pub(crate) fn create<'a>(
-        node_handle: Arc<NodeHandle>,
         options: impl Into<PublisherOptions<'a>>,
+        node_handle: Arc<NodeHandle>,
     ) -> Result<Arc<Self>, RclrsError>
     where
         T: Message,