From 74bffd29da142a77216440aea5882c066243efe3 Mon Sep 17 00:00:00 2001 From: Yuankun Zhu Date: Wed, 26 Aug 2026 10:23:51 -0700 Subject: [PATCH] feat: support native buffer subscriptions Expose acceptable buffer backend negotiation and align dynamic primitive sequence access with the ROS C message ABI. --- .../dynamic_message/dynamic_subscription.rs | 35 ++++- rclrs/src/dynamic_message/field_access.rs | 126 +++++++++--------- .../field_access/dynamic_sequence.rs | 126 +++++++++++++++++- rclrs/src/parameter/service.rs | 2 +- rclrs/src/subscription.rs | 63 ++++++++- 5 files changed, 276 insertions(+), 76 deletions(-) diff --git a/rclrs/src/dynamic_message/dynamic_subscription.rs b/rclrs/src/dynamic_message/dynamic_subscription.rs index ddc3de290..d94e97c6c 100644 --- a/rclrs/src/dynamic_message/dynamic_subscription.rs +++ b/rclrs/src/dynamic_message/dynamic_subscription.rs @@ -278,7 +278,9 @@ where ) -> Result, RclrsError> { // This loads the introspection type support library. let metadata = DynamicMessageMetadata::new(topic_type)?; - let SubscriptionOptions { topic, qos } = options.into(); + let options = options.into(); + let topic = options.topic; + let qos = options.qos; // However, we also need the regular type support library – // the rosidl_typesupport_c one. let message_type = &metadata.message_type; @@ -303,9 +305,30 @@ where // SAFETY: No preconditions for this function. let mut rcl_subscription_options = unsafe { rcl_subscription_get_default_options() }; rcl_subscription_options.qos = qos.into(); + #[cfg(ros_distro = "rolling")] + if let Some(backends) = options.acceptable_buffer_backends { + let backends_c_string = + CString::new(backends).map_err(|err| RclrsError::StringContainsNul { + err, + s: backends.into(), + })?; + let set_result = unsafe { + rcl_subscription_options_set_acceptable_buffer_backends( + backends_c_string.as_ptr(), + &mut rcl_subscription_options, + ) + .ok() + }; + if let Err(error) = set_result { + unsafe { + rcl_subscription_options_fini(&mut rcl_subscription_options); + } + return Err(error); + } + } // SAFETY: Getting a zero-initialized value is always safe. let mut rcl_subscription = unsafe { rcl_get_zero_initialized_subscription() }; - { + let init_result = { let rcl_node = node_handle.rcl_node.lock().unwrap(); let _lifecycle_lock = ENTITY_LIFECYCLE_MUTEX.lock().unwrap(); unsafe { @@ -322,9 +345,13 @@ where topic_c_string.as_ptr(), &rcl_subscription_options, ) - .ok()?; + .ok() } - } + }; + let options_fini_result = + unsafe { rcl_subscription_options_fini(&mut rcl_subscription_options).ok() }; + init_result?; + options_fini_result?; let handle = Arc::new(SubscriptionHandle { rcl_subscription: Mutex::new(rcl_subscription), diff --git a/rclrs/src/dynamic_message/field_access.rs b/rclrs/src/dynamic_message/field_access.rs index bcddc0b1a..beab0504e 100644 --- a/rclrs/src/dynamic_message/field_access.rs +++ b/rclrs/src/dynamic_message/field_access.rs @@ -1,4 +1,4 @@ -use rosidl_runtime_rs::Sequence; +use rosidl_runtime_rs::{PrimitiveSequence, Sequence}; use super::{BaseType, MessageFieldInfo, ValueKind}; @@ -122,8 +122,8 @@ macro_rules! define_value_types { #[allow(missing_docs)] #[derive(Debug, PartialEq)] pub enum SequenceValue<'msg> { - FloatSequence(make_ref!('msg, Sequence)), - DoubleSequence(make_ref!('msg, Sequence)), + FloatSequence(make_ref!('msg, PrimitiveSequence)), + DoubleSequence(make_ref!('msg, PrimitiveSequence)), /// It's platform-dependent what the size of long double is. /// Here's a pointer to the [`Sequence`][1] struct. /// @@ -132,18 +132,18 @@ macro_rules! define_value_types { immutable => *const u8, mutable => *mut u8 )), - CharSequence(make_ref!('msg, Sequence)), - WCharSequence(make_ref!('msg, Sequence)), - BooleanSequence(make_ref!('msg, Sequence)), - OctetSequence(make_ref!('msg, Sequence)), - Uint8Sequence(make_ref!('msg, Sequence)), - Int8Sequence(make_ref!('msg, Sequence)), - Uint16Sequence(make_ref!('msg, Sequence)), - Int16Sequence(make_ref!('msg, Sequence)), - Uint32Sequence(make_ref!('msg, Sequence)), - Int32Sequence(make_ref!('msg, Sequence)), - Uint64Sequence(make_ref!('msg, Sequence)), - Int64Sequence(make_ref!('msg, Sequence)), + CharSequence(make_ref!('msg, PrimitiveSequence)), + WCharSequence(make_ref!('msg, PrimitiveSequence)), + BooleanSequence(make_ref!('msg, PrimitiveSequence)), + OctetSequence(make_ref!('msg, PrimitiveSequence)), + Uint8Sequence(make_ref!('msg, PrimitiveSequence)), + Int8Sequence(make_ref!('msg, PrimitiveSequence)), + Uint16Sequence(make_ref!('msg, PrimitiveSequence)), + Int16Sequence(make_ref!('msg, PrimitiveSequence)), + Uint32Sequence(make_ref!('msg, PrimitiveSequence)), + Int32Sequence(make_ref!('msg, PrimitiveSequence)), + Uint64Sequence(make_ref!('msg, PrimitiveSequence)), + Int64Sequence(make_ref!('msg, PrimitiveSequence)), StringSequence(make_ref!('msg, Sequence)), /// This variant is not a [`Sequence`][1], since there is no suitable element type /// that both matches the underlying struct layout and includes information about @@ -180,6 +180,10 @@ macro_rules! define_value_types { immutable => DynamicBoundedSequence<'msg, T>, mutable => DynamicBoundedSequenceMut<'msg, T> ); + type BoundedPrimitiveSequence<'msg, T> = $select!( + immutable => DynamicBoundedSequence<'msg, T>, + mutable => DynamicBoundedPrimitiveSequenceMut<'msg, T> + ); /// A sequence of bounded length. // The field variants are for the most part self-explaining. @@ -187,8 +191,8 @@ macro_rules! define_value_types { #[allow(missing_docs)] #[derive(Debug, PartialEq)] pub enum BoundedSequenceValue<'msg> { - FloatBoundedSequence(BoundedSequence<'msg, f32>), - DoubleBoundedSequence(BoundedSequence<'msg, f64>), + FloatBoundedSequence(BoundedPrimitiveSequence<'msg, f32>), + DoubleBoundedSequence(BoundedPrimitiveSequence<'msg, f64>), /// It's platform-dependent what the size of long double is. /// Here's a pointer to the [`BoundedSequence`][1] struct and the upper bound. /// @@ -197,18 +201,18 @@ macro_rules! define_value_types { immutable => *const u8, mutable => *mut u8 ), usize), - CharBoundedSequence(BoundedSequence<'msg, u8>), - WCharBoundedSequence(BoundedSequence<'msg, u16>), - BooleanBoundedSequence(BoundedSequence<'msg, bool>), - OctetBoundedSequence(BoundedSequence<'msg, u8>), - Uint8BoundedSequence(BoundedSequence<'msg, u8>), - Int8BoundedSequence(BoundedSequence<'msg, i8>), - Uint16BoundedSequence(BoundedSequence<'msg, u16>), - Int16BoundedSequence(BoundedSequence<'msg, i16>), - Uint32BoundedSequence(BoundedSequence<'msg, u32>), - Int32BoundedSequence(BoundedSequence<'msg, i32>), - Uint64BoundedSequence(BoundedSequence<'msg, u64>), - Int64BoundedSequence(BoundedSequence<'msg, i64>), + CharBoundedSequence(BoundedPrimitiveSequence<'msg, u8>), + WCharBoundedSequence(BoundedPrimitiveSequence<'msg, u16>), + BooleanBoundedSequence(BoundedPrimitiveSequence<'msg, bool>), + OctetBoundedSequence(BoundedPrimitiveSequence<'msg, u8>), + Uint8BoundedSequence(BoundedPrimitiveSequence<'msg, u8>), + Int8BoundedSequence(BoundedPrimitiveSequence<'msg, i8>), + Uint16BoundedSequence(BoundedPrimitiveSequence<'msg, u16>), + Int16BoundedSequence(BoundedPrimitiveSequence<'msg, i16>), + Uint32BoundedSequence(BoundedPrimitiveSequence<'msg, u32>), + Int32BoundedSequence(BoundedPrimitiveSequence<'msg, i32>), + Uint64BoundedSequence(BoundedPrimitiveSequence<'msg, u64>), + Int64BoundedSequence(BoundedPrimitiveSequence<'msg, i64>), StringBoundedSequence(BoundedSequence<'msg, rosidl_runtime_rs::String>), BoundedStringBoundedSequence($select!( immutable => DynamicBoundedSequence<'msg, DynamicBoundedString<'msg>>, @@ -473,50 +477,50 @@ macro_rules! define_value_types { ) -> Self { match &field_info.base_type { BaseType::Float => { - SequenceValue::FloatSequence(reinterpret::>(bytes)) + SequenceValue::FloatSequence(reinterpret::>(bytes)) } BaseType::Double => { - SequenceValue::DoubleSequence(reinterpret::>(bytes)) + SequenceValue::DoubleSequence(reinterpret::>(bytes)) } BaseType::LongDouble => SequenceValue::LongDoubleSequence($select!( immutable => bytes.as_ptr(), mutable => bytes.as_mut_ptr() )), BaseType::Char => { - SequenceValue::CharSequence(reinterpret::>(bytes)) + SequenceValue::CharSequence(reinterpret::>(bytes)) } BaseType::WChar => { - SequenceValue::WCharSequence(reinterpret::>(bytes)) + SequenceValue::WCharSequence(reinterpret::>(bytes)) } BaseType::Boolean => { - SequenceValue::BooleanSequence(reinterpret::>(bytes)) + SequenceValue::BooleanSequence(reinterpret::>(bytes)) } BaseType::Octet => { - SequenceValue::OctetSequence(reinterpret::>(bytes)) + SequenceValue::OctetSequence(reinterpret::>(bytes)) } BaseType::Uint8 => { - SequenceValue::Uint8Sequence(reinterpret::>(bytes)) + SequenceValue::Uint8Sequence(reinterpret::>(bytes)) } BaseType::Int8 => { - SequenceValue::Int8Sequence(reinterpret::>(bytes)) + SequenceValue::Int8Sequence(reinterpret::>(bytes)) } BaseType::Uint16 => { - SequenceValue::Uint16Sequence(reinterpret::>(bytes)) + SequenceValue::Uint16Sequence(reinterpret::>(bytes)) } BaseType::Int16 => { - SequenceValue::Int16Sequence(reinterpret::>(bytes)) + SequenceValue::Int16Sequence(reinterpret::>(bytes)) } BaseType::Uint32 => { - SequenceValue::Uint32Sequence(reinterpret::>(bytes)) + SequenceValue::Uint32Sequence(reinterpret::>(bytes)) } BaseType::Int32 => { - SequenceValue::Int32Sequence(reinterpret::>(bytes)) + SequenceValue::Int32Sequence(reinterpret::>(bytes)) } BaseType::Uint64 => { - SequenceValue::Uint64Sequence(reinterpret::>(bytes)) + SequenceValue::Uint64Sequence(reinterpret::>(bytes)) } BaseType::Int64 => { - SequenceValue::Int64Sequence(reinterpret::>(bytes)) + SequenceValue::Int64Sequence(reinterpret::>(bytes)) } BaseType::String => { SequenceValue::StringSequence(reinterpret::< @@ -596,7 +600,7 @@ macro_rules! define_value_types { sequence_upper_bound ) }, - mutable => DynamicBoundedSequenceMut::new_primitive( + mutable => DynamicBoundedPrimitiveSequenceMut::new_primitive( bytes, sequence_upper_bound, field_info.resize_function.unwrap(), @@ -611,7 +615,7 @@ macro_rules! define_value_types { sequence_upper_bound ) }, - mutable => DynamicBoundedSequenceMut::new_primitive( + mutable => DynamicBoundedPrimitiveSequenceMut::new_primitive( bytes, sequence_upper_bound, field_info.resize_function.unwrap(), @@ -633,7 +637,7 @@ macro_rules! define_value_types { sequence_upper_bound ) }, - mutable => DynamicBoundedSequenceMut::new_primitive( + mutable => DynamicBoundedPrimitiveSequenceMut::new_primitive( bytes, sequence_upper_bound, field_info.resize_function.unwrap(), @@ -648,7 +652,7 @@ macro_rules! define_value_types { sequence_upper_bound ) }, - mutable => DynamicBoundedSequenceMut::new_primitive( + mutable => DynamicBoundedPrimitiveSequenceMut::new_primitive( bytes, sequence_upper_bound, field_info.resize_function.unwrap(), @@ -663,7 +667,7 @@ macro_rules! define_value_types { sequence_upper_bound ) }, - mutable => DynamicBoundedSequenceMut::new_primitive( + mutable => DynamicBoundedPrimitiveSequenceMut::new_primitive( bytes, sequence_upper_bound, field_info.resize_function.unwrap(), @@ -678,7 +682,7 @@ macro_rules! define_value_types { sequence_upper_bound ) }, - mutable => DynamicBoundedSequenceMut::new_primitive( + mutable => DynamicBoundedPrimitiveSequenceMut::new_primitive( bytes, sequence_upper_bound, field_info.resize_function.unwrap(), @@ -693,7 +697,7 @@ macro_rules! define_value_types { sequence_upper_bound ) }, - mutable => DynamicBoundedSequenceMut::new_primitive( + mutable => DynamicBoundedPrimitiveSequenceMut::new_primitive( bytes, sequence_upper_bound, field_info.resize_function.unwrap(), @@ -708,7 +712,7 @@ macro_rules! define_value_types { sequence_upper_bound ) }, - mutable => DynamicBoundedSequenceMut::new_primitive( + mutable => DynamicBoundedPrimitiveSequenceMut::new_primitive( bytes, sequence_upper_bound, field_info.resize_function.unwrap(), @@ -723,7 +727,7 @@ macro_rules! define_value_types { sequence_upper_bound ) }, - mutable => DynamicBoundedSequenceMut::new_primitive( + mutable => DynamicBoundedPrimitiveSequenceMut::new_primitive( bytes, sequence_upper_bound, field_info.resize_function.unwrap(), @@ -738,7 +742,7 @@ macro_rules! define_value_types { sequence_upper_bound ) }, - mutable => DynamicBoundedSequenceMut::new_primitive( + mutable => DynamicBoundedPrimitiveSequenceMut::new_primitive( bytes, sequence_upper_bound, field_info.resize_function.unwrap(), @@ -753,7 +757,7 @@ macro_rules! define_value_types { sequence_upper_bound ) }, - mutable => DynamicBoundedSequenceMut::new_primitive( + mutable => DynamicBoundedPrimitiveSequenceMut::new_primitive( bytes, sequence_upper_bound, field_info.resize_function.unwrap(), @@ -768,7 +772,7 @@ macro_rules! define_value_types { sequence_upper_bound ) }, - mutable => DynamicBoundedSequenceMut::new_primitive( + mutable => DynamicBoundedPrimitiveSequenceMut::new_primitive( bytes, sequence_upper_bound, field_info.resize_function.unwrap(), @@ -783,7 +787,7 @@ macro_rules! define_value_types { sequence_upper_bound ) }, - mutable => DynamicBoundedSequenceMut::new_primitive( + mutable => DynamicBoundedPrimitiveSequenceMut::new_primitive( bytes, sequence_upper_bound, field_info.resize_function.unwrap(), @@ -798,7 +802,7 @@ macro_rules! define_value_types { sequence_upper_bound ) }, - mutable => DynamicBoundedSequenceMut::new_primitive( + mutable => DynamicBoundedPrimitiveSequenceMut::new_primitive( bytes, sequence_upper_bound, field_info.resize_function.unwrap(), @@ -808,12 +812,12 @@ macro_rules! define_value_types { BaseType::String => { BoundedSequenceValue::StringBoundedSequence($select!( immutable => { - DynamicBoundedSequence::new_primitive( + DynamicBoundedSequence::new_native( bytes, sequence_upper_bound ) }, - mutable => DynamicBoundedSequenceMut::new_primitive( + mutable => DynamicBoundedSequenceMut::new_native( bytes, sequence_upper_bound, field_info.resize_function.unwrap() @@ -834,12 +838,12 @@ macro_rules! define_value_types { BaseType::WString => { BoundedSequenceValue::WStringBoundedSequence($select!( immutable => { - DynamicBoundedSequence::new_primitive( + DynamicBoundedSequence::new_native( bytes, sequence_upper_bound ) }, - mutable => DynamicBoundedSequenceMut::new_primitive( + mutable => DynamicBoundedSequenceMut::new_native( bytes, sequence_upper_bound, field_info.resize_function.unwrap() diff --git a/rclrs/src/dynamic_message/field_access/dynamic_sequence.rs b/rclrs/src/dynamic_message/field_access/dynamic_sequence.rs index e768f1fc6..88c275a8e 100644 --- a/rclrs/src/dynamic_message/field_access/dynamic_sequence.rs +++ b/rclrs/src/dynamic_message/field_access/dynamic_sequence.rs @@ -4,7 +4,9 @@ use std::{ ops::{Deref, DerefMut}, }; -use rosidl_runtime_rs::{Sequence, SequenceAlloc, SequenceExceedsBoundsError}; +use rosidl_runtime_rs::{ + PrimitiveSequence, PrimitiveSequenceAlloc, Sequence, SequenceAlloc, SequenceExceedsBoundsError, +}; use super::check; @@ -156,6 +158,24 @@ where } } +impl<'msg, T> InnerSequence for &'msg mut PrimitiveSequence +where + T: PartialEq + PrimitiveSequenceAlloc, +{ + fn as_slice(&self) -> &[T] { + PrimitiveSequence::as_slice(self) + } + + fn as_mut_slice(&mut self) -> &mut [T] { + PrimitiveSequence::as_mut_slice(self) + } + + fn resize_unchecked(&mut self, resize_function: ResizeFunction, len: usize) { + let is_ok = unsafe { resize_function(*self as *mut _ as *mut std::os::raw::c_void, len) }; + assert!(is_ok); + } +} + impl<'msg, T> PartialEq for ProxySequence<'msg, T> where T: PartialEq + ProxyMut<'msg>, @@ -331,9 +351,23 @@ impl<'msg, T> Deref for DynamicBoundedSequence<'msg, T> { impl<'msg, T> DynamicBoundedSequence<'msg, T> where - T: SequenceAlloc, + T: PrimitiveSequenceAlloc, { pub(super) unsafe fn new_primitive(bytes: &'msg [u8], upper_bound: usize) -> Self { + let sequence = &*(bytes.as_ptr() as *const PrimitiveSequence); + let slice = sequence.as_slice(); + Self { + boo: BooSlice::Borrowed(slice), + upper_bound, + } + } +} + +impl<'msg, T> DynamicBoundedSequence<'msg, T> +where + T: SequenceAlloc, +{ + pub(super) unsafe fn new_native(bytes: &'msg [u8], upper_bound: usize) -> Self { let sequence = &*(bytes.as_ptr() as *const Sequence); let slice = sequence.as_slice(); Self { @@ -360,7 +394,7 @@ where } } -impl<'msg, T: SequenceAlloc> DynamicBoundedSequence<'msg, T> { +impl<'msg, T> DynamicBoundedSequence<'msg, T> { /// See [`Sequence::as_slice()`][1]. /// /// [1]: rosidl_runtime_rs::Sequence::as_slice @@ -406,6 +440,86 @@ pub struct DynamicBoundedSequenceMut<'msg, T: DynamicSequenceElementMut<'msg>> { upper_bound: usize, } +/// A mutable bounded primitive sequence whose bound is known at runtime. +#[derive(PartialEq)] +pub struct DynamicBoundedPrimitiveSequenceMut<'msg, T: PrimitiveSequenceAlloc> { + sequence: &'msg mut PrimitiveSequence, + resize_function: ResizeFunction, + upper_bound: usize, +} + +impl Debug for DynamicBoundedPrimitiveSequenceMut<'_, T> +where + T: Debug + PrimitiveSequenceAlloc, +{ + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> Result<(), fmt::Error> { + self.sequence.as_slice().fmt(f) + } +} + +impl Deref for DynamicBoundedPrimitiveSequenceMut<'_, T> { + type Target = [T]; + + fn deref(&self) -> &Self::Target { + self.sequence.as_slice() + } +} + +impl DerefMut for DynamicBoundedPrimitiveSequenceMut<'_, T> { + fn deref_mut(&mut self) -> &mut Self::Target { + self.sequence.as_mut_slice() + } +} + +impl<'msg, T: PrimitiveSequenceAlloc> DynamicBoundedPrimitiveSequenceMut<'msg, T> { + pub(super) unsafe fn new_primitive( + bytes: &'msg mut [u8], + upper_bound: usize, + resize_function: ResizeFunction, + ) -> Self { + Self { + sequence: &mut *(bytes.as_mut_ptr() as *mut PrimitiveSequence), + resize_function, + upper_bound, + } + } + + /// Returns the maximum length of this sequence. + pub fn upper_bound(&self) -> usize { + self.upper_bound + } + + /// Returns the sequence elements as a slice. + pub fn as_slice(&self) -> &[T] { + self.sequence.as_slice() + } + + /// Returns the sequence elements as a mutable slice. + pub fn as_mut_slice(&mut self) -> &mut [T] { + self.sequence.as_mut_slice() + } + + /// Tries to reset this sequence to `len` zero-initialized elements. + pub fn try_reset(&mut self, len: usize) -> Result<(), SequenceExceedsBoundsError> { + if len > self.upper_bound { + return Err(SequenceExceedsBoundsError { + len, + upper_bound: self.upper_bound, + }); + } + let is_ok = unsafe { + (self.resize_function)(self.sequence as *mut _ as *mut std::os::raw::c_void, len) + }; + assert!(is_ok); + Ok(()) + } + + /// Resets this sequence to empty. + pub fn clear(&mut self) { + self.try_reset(0).unwrap(); + } +} + // ------------------------- impl for DynamicSequenceMut ------------------------- impl<'msg, T> Debug for DynamicSequenceMut<'msg, T> @@ -442,7 +556,7 @@ where + DynamicSequenceElementMut<'msg, InnerSequence = &'msg mut Sequence> + 'static, { - pub(super) unsafe fn new_primitive( + pub(super) unsafe fn new_native( bytes: &'msg mut [u8], resize_function: ResizeFunction, ) -> Self { @@ -538,12 +652,12 @@ where + DynamicSequenceElementMut<'msg, InnerSequence = &'msg mut Sequence> + 'static, { - pub(super) unsafe fn new_primitive( + pub(super) unsafe fn new_native( bytes: &'msg mut [u8], upper_bound: usize, resize_function: ResizeFunction, ) -> Self { - let inner = DynamicSequenceMut::new_primitive(bytes, resize_function); + let inner = DynamicSequenceMut::new_native(bytes, resize_function); Self { inner, upper_bound } } } diff --git a/rclrs/src/parameter/service.rs b/rclrs/src/parameter/service.rs index 79ea85e96..5dc8dc63e 100644 --- a/rclrs/src/parameter/service.rs +++ b/rclrs/src/parameter/service.rs @@ -354,7 +354,7 @@ mod tests { Ok(()) } } - use rosidl_runtime_rs::{seq, Sequence}; + use rosidl_runtime_rs::seq; use std::{ sync::{ atomic::{AtomicBool, Ordering}, diff --git a/rclrs/src/subscription.rs b/rclrs/src/subscription.rs index 16a800a9c..9f74ab119 100644 --- a/rclrs/src/subscription.rs +++ b/rclrs/src/subscription.rs @@ -131,7 +131,9 @@ where node_handle: &Arc, commands: &Arc, ) -> Result, RclrsError> { - let SubscriptionOptions { topic, qos } = options.into(); + let options = options.into(); + let topic = options.topic; + let qos = options.qos; let callback = Arc::new(Mutex::new(callback)); // SAFETY: Getting a zero-initialized value is always safe. @@ -146,8 +148,29 @@ where // SAFETY: No preconditions for this function. let mut rcl_subscription_options = unsafe { rcl_subscription_get_default_options() }; rcl_subscription_options.qos = qos.into(); + #[cfg(ros_distro = "rolling")] + if let Some(backends) = options.acceptable_buffer_backends { + let backends_c_string = + CString::new(backends).map_err(|err| RclrsError::StringContainsNul { + err, + s: backends.into(), + })?; + let set_result = unsafe { + rcl_subscription_options_set_acceptable_buffer_backends( + backends_c_string.as_ptr(), + &mut rcl_subscription_options, + ) + .ok() + }; + if let Err(error) = set_result { + unsafe { + rcl_subscription_options_fini(&mut rcl_subscription_options); + } + return Err(error); + } + } - { + let init_result = { let rcl_node = node_handle.rcl_node.lock().unwrap(); let _lifecycle_lock = ENTITY_LIFECYCLE_MUTEX.lock().unwrap(); unsafe { @@ -164,9 +187,13 @@ where topic_c_string.as_ptr(), &rcl_subscription_options, ) - .ok()?; + .ok() } - } + }; + let options_fini_result = + unsafe { rcl_subscription_options_fini(&mut rcl_subscription_options).ok() }; + init_result?; + options_fini_result?; let handle = Arc::new(SubscriptionHandle { rcl_subscription: Mutex::new(rcl_subscription), @@ -244,6 +271,12 @@ pub struct SubscriptionOptions<'a> { pub topic: &'a str, /// The quality of service settings for the subscription. pub qos: QoSProfile, + /// Buffer backends accepted by the subscription. + /// + /// `None`, an empty string, or `"cpu"` selects CPU buffers. `"any"` accepts + /// every installed backend. A comma-separated list selects specific backends. + #[cfg(ros_distro = "rolling")] + pub acceptable_buffer_backends: Option<&'a str>, } impl<'a> SubscriptionOptions<'a> { @@ -252,8 +285,17 @@ impl<'a> SubscriptionOptions<'a> { Self { topic, qos: QoSProfile::topics_default(), + #[cfg(ros_distro = "rolling")] + acceptable_buffer_backends: None, } } + + /// Sets the Buffer backends accepted by this subscription. + #[cfg(ros_distro = "rolling")] + pub fn acceptable_buffer_backends(mut self, backends: &'a str) -> Self { + self.acceptable_buffer_backends = Some(backends); + self + } } impl<'a, T: IntoPrimitiveOptions<'a>> From for SubscriptionOptions<'a> { @@ -660,6 +702,8 @@ mod tests { SubscriptionOptions { topic: "test_subscription_qos_topic_3", qos: expected_qos, + #[cfg(ros_distro = "rolling")] + acceptable_buffer_backends: None, }, |_: Empty| { // Do nothing @@ -670,6 +714,17 @@ mod tests { let qos = subscription.qos(); assert_eq!(expected_qos.reliability, qos.reliability); assert_eq!(qos.reliability, QoSReliabilityPolicy::BestEffort); + + #[cfg(ros_distro = "rolling")] + node.create_subscription( + SubscriptionOptions { + topic: "test_subscription_cuda_buffer_backend", + qos: QoSProfile::topics_default(), + acceptable_buffer_backends: Some("cuda"), + }, + |_: Empty| {}, + ) + .unwrap(); } #[test]