Reworking Commands API

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

Author: mxgrey

rclrs/src/executor.rs

@@ -122,7 +122,7 @@ impl Executor {
 /// while the executor is spinning.
 pub struct ExecutorCommands {
     context: Context,
-    channel: Box<dyn ExecutorChannel>,
+    channel: Arc<dyn ExecutorChannel>,
     halt_spinning: Arc<AtomicBool>,
     wakeup_wait_set: Arc<GuardCondition>,
 }
@@ -216,31 +216,60 @@ impl ExecutorCommands {
     pub(crate) fn get_guard_condition(&self) -> &Arc<GuardCondition> {
         &self.wakeup_wait_set
     }
+
+    pub(crate) fn as_worker_commands(&self) -> WorkerCommands {
+        WorkerCommands::new(Arc::clone(&self.channel).as_worker_channel())
+    }
+}
+
+#[derive(Clone)]
+pub(crate) struct WorkerCommands {
+    channel: Arc<dyn WorkerChannel>,
+}
+
+impl WorkerCommands {
+    pub(crate) fn new(channel: Arc<dyn WorkerChannel>) -> Self {
+        Self { channel }
+    }
+
+    pub(crate) fn add_to_wait_set(&self, waitable: Waitable) {
+        self.channel.add_to_waitset(waitable);
+    }
+
+    pub(crate) fn run<F>(&self, f: F)
+    where
+        F: 'static + Future<Output = ()> + Send,
+    {
+        self.channel.add_async_task(Box::pin(f));
+    }
 }
 
 pub trait WorkerChannel: Send + Sync {
+    /// Add a new item for the executor to run.
+    fn add_async_task(&self, f: BoxFuture<'static, ()>);
+
     /// Add new entities to the waitset of the executor.
     fn add_to_waitset(&self, new_entity: Waitable);
+
+    /// Allows an ExecutorChannel to be cast to a WorkerChannel.
+    fn as_worker_channel(self: Arc<Self>) -> Arc<dyn WorkerChannel + 'static>;
 }
 
 /// This trait defines the interface for passing new items into an executor to
 /// run.
 pub trait ExecutorChannel: WorkerChannel {
-    /// Add a new item for the executor to run.
-    fn add_async_task(&self, f: BoxFuture<'static, ()>);
-
     /// Create a new channel specific to a worker whose payload must be
     /// initialized with the given function.
-    fn create_worker_channel(
+    fn create_worker(
         &self,
         payload: Box<dyn Any + Send>,
-    ) -> Box<dyn ExecutorChannel>;
+    ) -> Arc<dyn WorkerChannel>;
 }
 
 /// This trait defines the interface for having an executor run.
 pub trait ExecutorRuntime: Send {
     /// Get a channel that can add new items for the executor to run.
-    fn default_channel(&self) -> Box<dyn ExecutorChannel>;
+    fn default_channel(&self) -> Arc<dyn ExecutorChannel>;
 
     /// Tell the runtime to spin while blocking any further execution until the
     /// spinning is complete.

rclrs/src/subscription.rs

@@ -7,7 +7,7 @@ use std::{
 use rosidl_runtime_rs::{Message, RmwMessage};
 
 use crate::{
-    error::ToResult, qos::QoSProfile, rcl_bindings::*, ExecutorCommands, IntoPrimitiveOptions,
+    error::ToResult, qos::QoSProfile, rcl_bindings::*, WorkerCommands, IntoPrimitiveOptions,
     Node, NodeHandle, RclPrimitive, RclPrimitiveHandle, RclPrimitiveKind, RclrsError, Waitable,
     WaitableLifecycle, Worker, ENTITY_LIFECYCLE_MUTEX,
 };
@@ -121,7 +121,7 @@ where
         options: impl Into<SubscriptionOptions<'a>>,
         callback: AnySubscriptionCallback<T, Scope::Payload>,
         node_handle: &Arc<NodeHandle>,
-        commands: &Arc<ExecutorCommands>,
+        commands: WorkerCommands,
     ) -> Result<Arc<Self>, RclrsError> {
         let SubscriptionOptions { topic, qos } = options.into();
         let callback = Arc::new(Mutex::new(callback));
@@ -169,7 +169,7 @@ where
             Box::new(SubscriptionExecutable {
                 handle: Arc::clone(&handle),
                 callback: Arc::clone(&callback),
-                commands: Arc::clone(commands),
+                commands: commands.clone(),
             }),
             Some(Arc::clone(commands.get_guard_condition())),
         );
@@ -218,7 +218,7 @@ impl<'a, T: IntoPrimitiveOptions<'a>> From<T> for SubscriptionOptions<'a> {
 struct SubscriptionExecutable<T: Message, Payload> {
     handle: Arc<SubscriptionHandle>,
     callback: Arc<Mutex<AnySubscriptionCallback<T, Payload>>>,
-    commands: Arc<ExecutorCommands>,
+    commands: WorkerCommands,
 }
 
 impl<T, Payload: 'static> RclPrimitive for SubscriptionExecutable<T, Payload>

rclrs/src/subscription/any_subscription_callback.rs

@@ -2,7 +2,7 @@ use rosidl_runtime_rs::Message;
 
 use crate::{
     subscription::SubscriptionHandle,
-    error::ToResult, rcl_bindings::*, ExecutorCommands, RclReturnCode, RclrsError,
+    error::ToResult, rcl_bindings::*, WorkerCommands, RclReturnCode, RclrsError,
     ReadOnlyLoanedMessage, MessageInfo, NodeSubscriptionCallback, WorkerSubscriptionCallback,
 };
 
@@ -21,7 +21,7 @@ impl<T: Message, Payload: 'static> AnySubscriptionCallback<T, Payload> {
         &mut self,
         handle: &Arc<SubscriptionHandle>,
         payload: &mut dyn Any,
-        commands: &Arc<ExecutorCommands>,
+        commands: &WorkerCommands,
     ) -> Result<(), RclrsError> {
         match self {
             Self::Node(node) => node.execute(handle, commands),

rclrs/src/subscription/node_subscription_callback.rs

@@ -1,10 +1,7 @@
 use rosidl_runtime_rs::Message;
 
 use super::{MessageInfo, SubscriptionHandle};
-use crate::{
-    ExecutorCommands, RclReturnCode, RclrsError,
-    ReadOnlyLoanedMessage, RclrsErrorFilter,
-};
+use crate::{WorkerCommands, RclrsError, ReadOnlyLoanedMessage, RclrsErrorFilter};
 
 use futures::future::BoxFuture;
 
@@ -40,33 +37,33 @@ impl<T: Message> NodeSubscriptionCallback<T> {
     pub(super) fn execute(
         &mut self,
         handle: &Arc<SubscriptionHandle>,
-        commands: &Arc<ExecutorCommands>,
+        commands: &WorkerCommands,
     ) -> Result<(), RclrsError> {
         let mut evaluate = || {
             match self {
                 NodeSubscriptionCallback::Regular(cb) => {
                     let (msg, _) = handle.take::<T>()?;
-                    let _ = commands.run(cb(msg));
+                    commands.run(cb(msg));
                 }
                 NodeSubscriptionCallback::RegularWithMessageInfo(cb) => {
                     let (msg, msg_info) = handle.take::<T>()?;
-                    let _ = commands.run(cb(msg, msg_info));
+                    commands.run(cb(msg, msg_info));
                 }
                 NodeSubscriptionCallback::Boxed(cb) => {
                     let (msg, _) = handle.take_boxed::<T>()?;
-                    let _ = commands.run(cb(msg));
+                    commands.run(cb(msg));
                 }
                 NodeSubscriptionCallback::BoxedWithMessageInfo(cb) => {
                     let (msg, msg_info) = handle.take_boxed::<T>()?;
-                    let _ = commands.run(cb(msg, msg_info));
+                    commands.run(cb(msg, msg_info));
                 }
                 NodeSubscriptionCallback::Loaned(cb) => {
                     let (msg, _) = handle.take_loaned::<T>()?;
-                    let _ = commands.run(cb(msg));
+                    commands.run(cb(msg));
                 }
                 NodeSubscriptionCallback::LoanedWithMessageInfo(cb) => {
                     let (msg, msg_info) = handle.take_loaned::<T>()?;
-                    let _ = commands.run(cb(msg, msg_info));
+                    commands.run(cb(msg, msg_info));
                 }
             }
             Ok(())