Extend SubscriptionCallback trait to ReadOnlyLoanedMessage (#266)

This was not a straightforward extension because
1. ReadOnlyLoanedMessage has a lifetime
2. The take_loaned_message() method was restricted to Subscription<T: RmwMessage>, i.e. not available
   for subscriptions on idiomatic messages.
   This has been fixed and now the method is implemented for Subscription<T: Message> like the other
   `take*` methods – though you still will get an RMW-native loaned message even if T is the idiomatic
   message type, because idiomatic message types can never be loaned.

Author: nnmm

examples/minimal_pub_sub/src/zero_copy_subscriber.rs

@@ -12,7 +12,7 @@ fn main() -> Result<(), Error> {
     let _subscription = node.create_subscription::<std_msgs::msg::UInt32, _>(
         "topic",
         rclrs::QOS_PROFILE_DEFAULT,
-        move |msg: std_msgs::msg::UInt32| {
+        move |msg: rclrs::ReadOnlyLoanedMessage<'_, std_msgs::msg::UInt32>| {
             num_messages += 1;
             println!("I heard: '{}'", msg.data);
             println!("(Got {} messages so far)", num_messages);

rclrs/src/publisher.rs

@@ -179,12 +179,13 @@ where
     /// 1. Publishers and subscriptions are on the same machine
     /// 1. The message is a "plain old data" type containing no variable-size members, whether bounded or unbounded
     /// 1. The publisher's QOS settings are compatible with zero-copy, e.g. the [default QOS][2]
-    /// 1. `Publisher::borrow_loaned_message()` and [`Subscription::take_loaned_message()`][1] are used
+    /// 1. `Publisher::borrow_loaned_message()` is used and the subscription uses a callback taking a
+    ///    [`ReadOnlyLoanedMessage`][1]
     ///
     /// This function is only implemented for [`RmwMessage`]s since the "idiomatic" message type
     /// does not have a typesupport library.
     ///
-    /// [1]: crate::Subscription::take_loaned_message
+    /// [1]: crate::ReadOnlyLoanedMessage
     /// [2]: crate::QOS_PROFILE_DEFAULT
     //
     // TODO: Explain more, e.g.

rclrs/src/subscription.rs

@@ -212,12 +212,7 @@ where
         };
         Ok(MessageInfo::from_rmw_message_info(&message_info))
     }
-}
 
-impl<T> Subscription<T>
-where
-    T: RmwMessage,
-{
     /// Obtains a read-only handle to a message owned by the middleware.
     ///
     /// When there is no new message, this will return a
@@ -228,23 +223,28 @@ where
     ///
     /// [1]: crate::RclrsError
     /// [2]: crate::Publisher::borrow_loaned_message
-    pub fn take_loaned_message(&self) -> Result<ReadOnlyLoanedMessage<'_, T>, RclrsError> {
+    pub fn take_loaned(&self) -> Result<(ReadOnlyLoanedMessage<'_, T>, MessageInfo), RclrsError> {
         let mut msg_ptr = std::ptr::null_mut();
+        let mut message_info = unsafe { rmw_get_zero_initialized_message_info() };
         unsafe {
             // SAFETY: The third argument (message_info) and fourth argument (allocation) may be null.
             // The second argument (loaned_message) contains a null ptr as expected.
             rcl_take_loaned_message(
                 &*self.handle.lock(),
                 &mut msg_ptr,
-                std::ptr::null_mut(),
+                &mut message_info,
                 std::ptr::null_mut(),
             )
             .ok()?;
         }
-        Ok(ReadOnlyLoanedMessage {
-            msg_ptr: msg_ptr as *const T,
+        let read_only_loaned_msg = ReadOnlyLoanedMessage {
+            msg_ptr: msg_ptr as *const T::RmwMsg,
             subscription: self,
-        })
+        };
+        Ok((
+            read_only_loaned_msg,
+            MessageInfo::from_rmw_message_info(&message_info),
+        ))
     }
 }
 
@@ -265,18 +265,26 @@ where
                     let (msg, _) = self.take()?;
                     cb(msg)
                 }
-                AnySubscriptionCallback::Boxed(cb) => {
-                    let (msg, _) = self.take_boxed()?;
-                    cb(msg)
-                }
                 AnySubscriptionCallback::RegularWithMessageInfo(cb) => {
                     let (msg, msg_info) = self.take()?;
                     cb(msg, msg_info)
                 }
+                AnySubscriptionCallback::Boxed(cb) => {
+                    let (msg, _) = self.take_boxed()?;
+                    cb(msg)
+                }
                 AnySubscriptionCallback::BoxedWithMessageInfo(cb) => {
                     let (msg, msg_info) = self.take_boxed()?;
                     cb(msg, msg_info)
                 }
+                AnySubscriptionCallback::Loaned(cb) => {
+                    let (msg, _) = self.take_loaned()?;
+                    cb(msg)
+                }
+                AnySubscriptionCallback::LoanedWithMessageInfo(cb) => {
+                    let (msg, msg_info) = self.take_loaned()?;
+                    cb(msg, msg_info)
+                }
             }
             Ok(())
         })() {

rclrs/src/subscription/callback.rs

@@ -1,9 +1,15 @@
 use super::MessageInfo;
+use crate::ReadOnlyLoanedMessage;
+
+use rosidl_runtime_rs::Message;
 
 /// A trait for allowed callbacks for subscriptions.
 ///
 /// See [`AnySubscriptionCallback`] for a list of possible callback signatures.
-pub trait SubscriptionCallback<T, Args>: Send {
+pub trait SubscriptionCallback<T, Args>: Send + 'static
+where
+    T: Message,
+{
     /// Converts the callback into an enum.
     ///
     /// User code never needs to call this function.
@@ -13,7 +19,10 @@ pub trait SubscriptionCallback<T, Args>: Send {
 /// An enum capturing the various possible function signatures for subscription callbacks.
 ///
 /// The correct enum variant is deduced by the [`SubscriptionCallback`] trait.
-pub enum AnySubscriptionCallback<T> {
+pub enum AnySubscriptionCallback<T>
+where
+    T: Message,
+{
     /// A callback with only the message as an argument.
     Regular(Box<dyn FnMut(T) + Send>),
     /// A callback with the message and the message info as arguments.
@@ -22,92 +31,104 @@ pub enum AnySubscriptionCallback<T> {
     Boxed(Box<dyn FnMut(Box<T>) + Send>),
     /// A callback with the boxed message and the message info as arguments.
     BoxedWithMessageInfo(Box<dyn FnMut(Box<T>, MessageInfo) + Send>),
+    /// A callback with only the loaned message as an argument.
+    #[allow(clippy::type_complexity)]
+    Loaned(Box<dyn for<'a> FnMut(ReadOnlyLoanedMessage<'a, T>) + Send>),
+    /// A callback with the loaned message and the message info as arguments.
+    #[allow(clippy::type_complexity)]
+    LoanedWithMessageInfo(Box<dyn for<'a> FnMut(ReadOnlyLoanedMessage<'a, T>, MessageInfo) + Send>),
 }
 
 // We need one implementation per arity. This was inspired by Bevy's systems.
 impl<T, A0, Func> SubscriptionCallback<T, (A0,)> for Func
 where
     Func: FnMut(A0) + Send + 'static,
-    (A0,): ArgTuple<T, Func = Box<dyn FnMut(A0) + Send>>,
+    (A0,): ArgTuple<T, Func>,
+    T: Message,
 {
     fn into_callback(self) -> AnySubscriptionCallback<T> {
-        <(A0,) as ArgTuple<T>>::into_callback_with_args(Box::new(self))
+        <(A0,) as ArgTuple<T, Func>>::into_callback_with_args(self)
     }
 }
 
 impl<T, A0, A1, Func> SubscriptionCallback<T, (A0, A1)> for Func
 where
     Func: FnMut(A0, A1) + Send + 'static,
-    (A0, A1): ArgTuple<T, Func = Box<dyn FnMut(A0, A1) + Send>>,
+    (A0, A1): ArgTuple<T, Func>,
+    T: Message,
 {
     fn into_callback(self) -> AnySubscriptionCallback<T> {
-        <(A0, A1) as ArgTuple<T>>::into_callback_with_args(Box::new(self))
+        <(A0, A1) as ArgTuple<T, Func>>::into_callback_with_args(self)
     }
 }
 
 // Helper trait for SubscriptionCallback.
 //
 // For each tuple of args, it provides conversion from a function with
 // these args to the correct enum variant.
-trait ArgTuple<T> {
-    type Func;
-    fn into_callback_with_args(func: Self::Func) -> AnySubscriptionCallback<T>;
+trait ArgTuple<T, Func>
+where
+    T: Message,
+{
+    fn into_callback_with_args(func: Func) -> AnySubscriptionCallback<T>;
 }
 
-impl<T> ArgTuple<T> for (Box<T>,) {
-    type Func = Box<dyn FnMut(Box<T>) + Send>;
-    fn into_callback_with_args(func: Self::Func) -> AnySubscriptionCallback<T> {
-        AnySubscriptionCallback::Boxed(func)
+impl<T, Func> ArgTuple<T, Func> for (T,)
+where
+    T: Message,
+    Func: FnMut(T) + Send + 'static,
+{
+    fn into_callback_with_args(func: Func) -> AnySubscriptionCallback<T> {
+        AnySubscriptionCallback::Regular(Box::new(func))
     }
 }
 
-impl<T> ArgTuple<T> for (Box<T>, MessageInfo) {
-    type Func = Box<dyn FnMut(Box<T>, MessageInfo) + Send>;
-    fn into_callback_with_args(func: Self::Func) -> AnySubscriptionCallback<T> {
-        AnySubscriptionCallback::BoxedWithMessageInfo(func)
+impl<T, Func> ArgTuple<T, Func> for (T, MessageInfo)
+where
+    T: Message,
+    Func: FnMut(T, MessageInfo) + Send + 'static,
+{
+    fn into_callback_with_args(func: Func) -> AnySubscriptionCallback<T> {
+        AnySubscriptionCallback::RegularWithMessageInfo(Box::new(func))
     }
 }
 
-impl<T> ArgTuple<T> for (T,) {
-    type Func = Box<dyn FnMut(T) + Send>;
-    fn into_callback_with_args(func: Self::Func) -> AnySubscriptionCallback<T> {
-        AnySubscriptionCallback::Regular(func)
+impl<T, Func> ArgTuple<T, Func> for (Box<T>,)
+where
+    T: Message,
+    Func: FnMut(Box<T>) + Send + 'static,
+{
+    fn into_callback_with_args(func: Func) -> AnySubscriptionCallback<T> {
+        AnySubscriptionCallback::Boxed(Box::new(func))
     }
 }
 
-impl<T> ArgTuple<T> for (T, MessageInfo) {
-    type Func = Box<dyn FnMut(T, MessageInfo) + Send>;
-    fn into_callback_with_args(func: Self::Func) -> AnySubscriptionCallback<T> {
-        AnySubscriptionCallback::RegularWithMessageInfo(func)
+impl<T, Func> ArgTuple<T, Func> for (Box<T>, MessageInfo)
+where
+    T: Message,
+    Func: FnMut(Box<T>, MessageInfo) + Send + 'static,
+{
+    fn into_callback_with_args(func: Func) -> AnySubscriptionCallback<T> {
+        AnySubscriptionCallback::BoxedWithMessageInfo(Box::new(func))
     }
 }
 
-#[cfg(test)]
-mod tests {
-    use super::*;
+impl<T, Func> ArgTuple<T, Func> for (ReadOnlyLoanedMessage<'_, T>,)
+where
+    T: Message,
+    Func: for<'b> FnMut(ReadOnlyLoanedMessage<'b, T>) + Send + 'static,
+{
+    fn into_callback_with_args(func: Func) -> AnySubscriptionCallback<T> {
+        AnySubscriptionCallback::Loaned(Box::new(func))
+    }
+}
 
-    #[test]
-    fn callback_conversion() {
-        struct Message;
-        let cb = |_msg: Message| {};
-        assert!(matches!(
-            cb.into_callback(),
-            AnySubscriptionCallback::<Message>::Regular(_)
-        ));
-        let cb = |_msg: Message, _info: MessageInfo| {};
-        assert!(matches!(
-            cb.into_callback(),
-            AnySubscriptionCallback::<Message>::RegularWithMessageInfo(_)
-        ));
-        let cb = |_msg: Box<Message>| {};
-        assert!(matches!(
-            cb.into_callback(),
-            AnySubscriptionCallback::<Message>::Boxed(_)
-        ));
-        let cb = |_msg: Box<Message>, _info: MessageInfo| {};
-        assert!(matches!(
-            cb.into_callback(),
-            AnySubscriptionCallback::<Message>::BoxedWithMessageInfo(_)
-        ));
+impl<T, Func> ArgTuple<T, Func> for (ReadOnlyLoanedMessage<'_, T>, MessageInfo)
+where
+    T: Message,
+    Func: for<'b> FnMut(ReadOnlyLoanedMessage<'b, T>, MessageInfo) + Send + 'static,
+{
+    fn into_callback_with_args(func: Func) -> AnySubscriptionCallback<T> {
+        AnySubscriptionCallback::LoanedWithMessageInfo(Box::new(func))
     }
 }

rclrs/src/subscription/readonly_loaned_message.rs

@@ -1,38 +1,41 @@
 use crate::rcl_bindings::*;
 use crate::{Subscription, ToResult};
 
-use rosidl_runtime_rs::RmwMessage;
+use rosidl_runtime_rs::Message;
 
 use std::ops::Deref;
 
 /// A message that is owned by the middleware, loaned out for reading.
 ///
-/// It dereferences to a `&T`.
+/// It dereferences to a `&T::RmwMsg`. That is, if `T` is already an RMW-native
+/// message, it's the same as `&T`, and otherwise it's the corresponding RMW-native
+/// message.
 ///
-/// This type is returned by [`Subscription::take_loaned_message()`].
+/// This type is returned by [`Subscription::take_loaned()`] and may be used in
+/// subscription callbacks.
 ///
 /// The loan is returned by dropping the `ReadOnlyLoanedMessage`.
 pub struct ReadOnlyLoanedMessage<'a, T>
 where
-    T: RmwMessage,
+    T: Message,
 {
-    pub(super) msg_ptr: *const T,
+    pub(super) msg_ptr: *const T::RmwMsg,
     pub(super) subscription: &'a Subscription<T>,
 }
 
 impl<'a, T> Deref for ReadOnlyLoanedMessage<'a, T>
 where
-    T: RmwMessage,
+    T: Message,
 {
-    type Target = T;
+    type Target = T::RmwMsg;
     fn deref(&self) -> &Self::Target {
         unsafe { &*self.msg_ptr }
     }
 }
 
 impl<'a, T> Drop for ReadOnlyLoanedMessage<'a, T>
 where
-    T: RmwMessage,
+    T: Message,
 {
     fn drop(&mut self) {
         unsafe {
@@ -48,6 +51,6 @@ where
 
 // SAFETY: The functions accessing this type, including drop(), shouldn't care about the thread
 // they are running in. Therefore, this type can be safely sent to another thread.
-unsafe impl<'a, T> Send for ReadOnlyLoanedMessage<'a, T> where T: RmwMessage {}
+unsafe impl<'a, T> Send for ReadOnlyLoanedMessage<'a, T> where T: Message {}
 // SAFETY: This type has no interior mutability, in fact it has no mutability at all.
-unsafe impl<'a, T> Sync for ReadOnlyLoanedMessage<'a, T> where T: RmwMessage {}
+unsafe impl<'a, T> Sync for ReadOnlyLoanedMessage<'a, T> where T: Message {}

rclrs_tests/src/pub_sub_tests.rs

@@ -1,4 +1,7 @@
-use rclrs::{LoanedMessage, Publisher, ReadOnlyLoanedMessage, Subscription};
+use rclrs::{
+    AnySubscriptionCallback, LoanedMessage, MessageInfo, Publisher, ReadOnlyLoanedMessage,
+    Subscription, SubscriptionCallback,
+};
 
 fn assert_send<T: Send>() {}
 fn assert_sync<T: Sync>() {}
@@ -26,3 +29,38 @@ fn readonly_loaned_message_is_send_and_sync() {
     assert_send::<ReadOnlyLoanedMessage<test_msgs::msg::rmw::BoundedSequences>>();
     assert_sync::<ReadOnlyLoanedMessage<test_msgs::msg::rmw::BoundedSequences>>();
 }
+
+#[test]
+fn callback_conversion() {
+    type Message = test_msgs::msg::BoundedSequences;
+    let cb = |_msg: Message| {};
+    assert!(matches!(
+        cb.into_callback(),
+        AnySubscriptionCallback::<Message>::Regular(_)
+    ));
+    let cb = |_msg: Message, _info: MessageInfo| {};
+    assert!(matches!(
+        cb.into_callback(),
+        AnySubscriptionCallback::<Message>::RegularWithMessageInfo(_)
+    ));
+    let cb = |_msg: Box<Message>| {};
+    assert!(matches!(
+        cb.into_callback(),
+        AnySubscriptionCallback::<Message>::Boxed(_)
+    ));
+    let cb = |_msg: Box<Message>, _info: MessageInfo| {};
+    assert!(matches!(
+        cb.into_callback(),
+        AnySubscriptionCallback::<Message>::BoxedWithMessageInfo(_)
+    ));
+    let cb = |_msg: ReadOnlyLoanedMessage<'_, Message>| {};
+    assert!(matches!(
+        cb.into_callback(),
+        AnySubscriptionCallback::<Message>::Loaned(_)
+    ));
+    let cb = |_msg: ReadOnlyLoanedMessage<'_, Message>, _info: MessageInfo| {};
+    assert!(matches!(
+        cb.into_callback(),
+        AnySubscriptionCallback::<Message>::LoanedWithMessageInfo(_)
+    ));
+}