feat: add push-callback registration for rcl primitives (Tokio Executor 1/4)

rclrs/src/client.rs

@@ -2,7 +2,7 @@ use std::{
     any::Any,
     collections::HashMap,
     ffi::{CStr, CString},
-    sync::{Arc, Mutex, MutexGuard},
+    sync::{atomic::AtomicBool, Arc, Mutex, MutexGuard},
 };
 
 use rosidl_runtime_rs::Message;
@@ -354,6 +354,7 @@ where
         let handle = Arc::new(ClientHandle {
             rcl_client: Mutex::new(rcl_client),
             node: Arc::clone(&node),
+            on_ready_slot: AtomicBool::new(false),
         });
 
         let board = Arc::new(Mutex::new(ClientRequestBoard::new()));
@@ -468,6 +469,39 @@ where
     fn kind(&self) -> RclPrimitiveKind {
         RclPrimitiveKind::Client
     }
+
+    fn register_on_ready(
+        &self,
+        on_ready: Box<dyn Fn(ReadyKind, usize) + Send + Sync>,
+    ) -> Result<Option<Box<dyn crate::OnReadyHandle>>, RclrsError> {
+        // A client has a single readiness path; report it as `Basic`.
+        let on_ready = move |n| on_ready(ReadyKind::Basic, n);
+        let registration = crate::executor::event_callback::OnReadyRegistration::new(
+            Arc::clone(&self.handle),
+            set_client_on_new_response,
+            client_on_ready_slot,
+            Box::new(on_ready),
+        )?;
+        Ok(Some(Box::new(registration)))
+    }
+}
+
+/// Install (or, with a null callback/user_data, clear) the "on new response"
+/// push callback used by the event-driven executor. Encapsulates the client
+/// lock and the rcl call within this module.
+unsafe fn set_client_on_new_response(
+    handle: &ClientHandle,
+    callback: rcl_event_callback_t,
+    user_data: *const std::os::raw::c_void,
+) -> rcl_ret_t {
+    rcl_client_set_on_new_response_callback(&*handle.lock(), callback, user_data)
+}
+
+/// The guard for the client's single "on new response" callback slot, paired
+/// with [`set_client_on_new_response`] so a push registration can enforce that
+/// only one callback owns the slot at a time.
+fn client_on_ready_slot(handle: &ClientHandle) -> &AtomicBool {
+    &handle.on_ready_slot
 }
 
 type SequenceNumber = i64;
@@ -581,6 +615,12 @@ struct ClientHandle {
     /// We store the whole node here because we use some of its user-facing API
     /// in some of the Client methods.
     node: Node,
+    /// Guards the single rcl "on new response" callback slot so only one push
+    /// registration can own it at a time. Selected via [`client_on_ready_slot`];
+    /// see [`OnReadySlotFn`].
+    ///
+    /// [`OnReadySlotFn`]: crate::executor::event_callback::OnReadySlotFn
+    on_ready_slot: AtomicBool,
 }
 
 impl ClientHandle {

rclrs/src/dynamic_message/dynamic_subscription.rs

@@ -3,7 +3,7 @@ use std::{
     boxed::Box,
     ffi::CString,
     ops::{Deref, DerefMut},
-    sync::{Arc, Mutex},
+    sync::{atomic::AtomicBool, Arc, Mutex},
 };
 
 use futures::future::BoxFuture;
@@ -329,6 +329,7 @@ where
         let handle = Arc::new(SubscriptionHandle {
             rcl_subscription: Mutex::new(rcl_subscription),
             node_handle: Arc::clone(node_handle),
+            on_ready_slot: AtomicBool::new(false),
         });
 
         let callback = Arc::new(Mutex::new(callback.into()));

rclrs/src/error.rs

@@ -80,6 +80,16 @@ pub enum RclrsError {
     /// However, the implementation of rclrs automatically protects from all of
     /// these errors except memory allocation failure.
     GoalAcceptanceError,
+    /// A second push "on ready" callback was registered on an entity that
+    /// already has one active.
+    ///
+    /// rcl stores a single callback per entity slot, so an overlapping
+    /// registration would silently clobber the existing one, and later dropping
+    /// the earlier registration would clear the newer callback. rclrs entities
+    /// register their callback exactly once and never replace it, so the second
+    /// registration is rejected rather than allowed to corrupt the RAII
+    /// handle's semantics.
+    OnReadyAlreadyRegistered,
 }
 
 impl RclrsError {
@@ -168,6 +178,13 @@ impl Display for RclrsError {
                     "An error occurred while trying to accept an action server goal",
                 )
             }
+            RclrsError::OnReadyAlreadyRegistered => {
+                write!(
+                    f,
+                    "A push on-ready callback is already registered for this entity. \
+                    rcl allows only one callback per entity slot",
+                )
+            }
         }
     }
 }
@@ -216,6 +233,7 @@ impl Error for RclrsError {
             RclrsError::PoisonedMutex => None,
             RclrsError::InvalidReadyInformation { .. } => None,
             RclrsError::GoalAcceptanceError => None,
+            RclrsError::OnReadyAlreadyRegistered => None,
         }
     }
 }

rclrs/src/executor.rs

@@ -1,6 +1,8 @@
 mod basic_executor;
 pub use self::basic_executor::*;
 
+pub(crate) mod event_callback;
+
 use crate::{
     Context, ContextHandle, GuardCondition, IntoNodeOptions, Node, RclrsError, Waitable,
     WeakActivityListener,