14#ifndef NROS_RCLCPP_SUBSCRIPTION_HPP
15#define NROS_RCLCPP_SUBSCRIPTION_HPP
21#include "rclcpp/detail/traits.hpp"
23#include "rclcpp/detail/entity_name.hpp"
26#include "rclcpp/subscription_handle.hpp"
27#include "rclcpp/inplace_fn.hpp"
28#include "rclcpp/size_bound.hpp"
37#include "rclcpp/detail/std_detect.hpp"
39#include "rclcpp/detail/nros_cpp_ffi.h"
49#include "rclcpp/message_info.hpp"
64 const char* type_name,
const char* type_hash,
67 void* context,
size_t* out_handle_id,
68 const nros_cpp_subscription_options_t* options);
93 const nros_cpp_node_t* node,
const char* topic,
const char* type_name,
const char* type_hash,
95 size_t capture_len,
void (*capture_drop)(
void* capture),
size_t* out_handle_id,
96 const nros_cpp_subscription_options_t* options);
101 const uint8_t* attachment,
102 size_t attachment_len,
void* ctx);
105 const nros_cpp_node_t* node,
const char* topic,
const char* type_name,
const char* type_hash,
107 size_t* out_handle_id,
const nros_cpp_subscription_options_t* options);
116#if defined(NROS_SAFETY_E2E)
117typedef void (*nros_cpp_subscription_validated_callback_t)(
const uint8_t* data,
size_t len,
118 int64_t gap,
bool duplicate,
119 int8_t crc_valid,
void* ctx);
122 const nros_cpp_node_t* node,
const char* topic,
const char* type_name,
const char* type_hash,
123 nros_cpp_qos_t qos, nros_cpp_subscription_validated_callback_t callback,
void* context,
124 size_t* out_handle_id,
const nros_cpp_subscription_options_t* options);
235 size_t attachment_len);
239#if defined(NROS_SAFETY_E2E)
240 using TypedSubscriptionSafetyFn = void (*)(
const M& msg,
241 const ::rclcpp::IntegrityStatus& integrity);
288 ::memcpy(topic_name_, other.topic_name_,
sizeof(topic_name_));
289 user_fn_ = other.user_fn_;
290 user_fn_ctx_ = other.user_fn_ctx_;
291 user_ctx_ = other.user_ctx_;
292 user_fn_info_ = other.user_fn_info_;
293#if defined(NROS_SAFETY_E2E)
294 user_fn_safety_ = other.user_fn_safety_;
296 sched_handle_id_ = other.sched_handle_id_;
297 executor_ = other.executor_;
298 other.initialized_ =
false;
302 if (
this != &other) {
303 initialized_ = other.initialized_;
304 ::memcpy(topic_name_, other.topic_name_,
sizeof(topic_name_));
305 user_fn_ = other.user_fn_;
306 user_fn_ctx_ = other.user_fn_ctx_;
307 user_ctx_ = other.user_ctx_;
308 user_fn_info_ = other.user_fn_info_;
309#if defined(NROS_SAFETY_E2E)
310 user_fn_safety_ = other.user_fn_safety_;
312 sched_handle_id_ = other.sched_handle_id_;
313 executor_ = other.executor_;
314 other.initialized_ =
false;
360 if (!initialized_ || executor_ ==
nullptr ||
361 nros_cpp_subscription_get_actual_qos(
nullptr, executor_, sched_handle_id_, &f) != 0) {
362 return ::rclcpp::detail::qos_all_unknown();
364 return ::rclcpp::detail::qos_from_ffi(f);
371 friend class ::rclcpp::Node;
376 static void message_trampoline(
const uint8_t* data,
size_t len,
void* ctx) {
378 if (self ==
nullptr)
return;
380 if (M::ffi_deserialize(data, len, &msg) != 0)
return;
381 if (self->user_fn_ !=
nullptr) {
383 }
else if (self->user_fn_ctx_ !=
nullptr) {
384 self->user_fn_ctx_(msg, self->user_ctx_);
391 static void message_info_trampoline(
const uint8_t* data,
size_t len,
const uint8_t* attachment,
392 size_t attachment_len,
void* ctx) {
394 if (self ==
nullptr)
return;
396 if (M::ffi_deserialize(data, len, &msg) != 0)
return;
397 if (self->user_fn_info_ !=
nullptr) {
398 self->user_fn_info_(msg, attachment, attachment_len);
402#if defined(NROS_SAFETY_E2E)
408 static void message_safety_trampoline(
const uint8_t* data,
size_t len, int64_t gap,
409 bool duplicate, int8_t crc_valid,
void* ctx) {
411 if (self ==
nullptr)
return;
413 if (M::ffi_deserialize(data, len, &msg) != 0)
return;
414 if (self->user_fn_safety_ !=
nullptr) {
415 nros_cpp_integrity_status_t status;
417 status.duplicate = duplicate;
418 status.crc_valid = crc_valid;
419 self->user_fn_safety_(msg, status);
430 void store_topic_name(
const char* topic) {
432 while (topic[n] !=
'\0' && n + 1 <
sizeof(topic_name_)) {
433 topic_name_[n] = topic[n];
436 topic_name_[n] =
'\0';
449 size_t sched_handle_id_ =
static_cast<size_t>(-1);
460 void* executor_ =
nullptr;
465 void* user_ctx_ =
nullptr;
466#if defined(NROS_SAFETY_E2E)
469 TypedSubscriptionSafetyFn user_fn_safety_ =
nullptr;
487template <
typename M,
typename F,
typename>
489 F callback, const ::rclcpp::SubscriptionOptions& options) {
499 out.user_fn_ctx_ =
nullptr;
500 out.user_ctx_ =
nullptr;
502 uint8_t sched = (options.sched_context == ::rclcpp::SCHED_CONTEXT_UNSET)
504 :
static_cast<uint8_t
>(options.sched_context);
505 size_t handle =
static_cast<size_t>(-1);
508 nros_cpp_subscription_options_t ffi_options = nros_cpp_subscription_default_options();
509 ffi_options.sched_context = sched;
514 ffi_options.rx_buffer_hint =
static_cast<uint32_t
>(::rclcpp::rx_buffer_capacity<M>::value);
515 ffi_options.stamp_offset = ::rclcpp::stamp_offset<M>::value;
517 &handle_, topic, M::TYPE_NAME, M::TYPE_HASH, ffi_qos, &Subscription<M>::message_trampoline,
518 &out, &
handle, &ffi_options);
520 out.sched_handle_id_ =
handle;
521 out.executor_ = executor_handle_;
524 out.store_topic_name(topic);
525 out.initialized_ =
true;
539template <
typename M,
typename F,
typename>
542 const ::rclcpp::QoS& qos, F callback,
543 const ::rclcpp::SubscriptionOptions& options) {
551 out.user_fn_ctx_ =
nullptr;
552 out.user_ctx_ =
nullptr;
554 uint8_t sched = (options.sched_context == ::rclcpp::SCHED_CONTEXT_UNSET)
556 :
static_cast<uint8_t
>(options.sched_context);
557 size_t handle =
static_cast<size_t>(-1);
559 nros_cpp_subscription_options_t ffi_options = nros_cpp_subscription_default_options();
560 ffi_options.sched_context = sched;
561 ffi_options.callback_group = group.get_name();
564 ffi_options.rx_buffer_hint =
static_cast<uint32_t
>(::rclcpp::rx_buffer_capacity<M>::value);
565 ffi_options.stamp_offset = ::rclcpp::stamp_offset<M>::value;
567 &handle_, topic, M::TYPE_NAME, M::TYPE_HASH, ffi_qos, &Subscription<M>::message_trampoline,
568 &out, &
handle, &ffi_options);
570 out.sched_handle_id_ =
handle;
571 out.executor_ = executor_handle_;
574 out.store_topic_name(topic);
575 out.initialized_ =
true;
590template <
typename M,
typename F,
typename>
592 const ::rclcpp::QoS& qos, F callback,
593 const ::rclcpp::SubscriptionOptions& options) {
601 out.user_fn_ =
nullptr;
602 out.user_fn_ctx_ =
nullptr;
603 out.user_ctx_ =
nullptr;
605 uint8_t sched = (options.sched_context == ::rclcpp::SCHED_CONTEXT_UNSET)
607 :
static_cast<uint8_t
>(options.sched_context);
608 size_t handle =
static_cast<size_t>(-1);
610 nros_cpp_subscription_options_t ffi_options = nros_cpp_subscription_default_options();
611 ffi_options.sched_context = sched;
615 ffi_options.rx_buffer_hint =
static_cast<uint32_t
>(::rclcpp::rx_buffer_capacity<M>::value);
616 ffi_options.stamp_offset = ::rclcpp::stamp_offset<M>::value;
618 &handle_, topic, M::TYPE_NAME, M::TYPE_HASH, ffi_qos,
619 &Subscription<M>::message_info_trampoline, &out, &
handle, &ffi_options);
621 out.sched_handle_id_ =
handle;
622 out.executor_ = executor_handle_;
625 out.store_topic_name(topic);
626 out.initialized_ =
true;
634#if defined(NROS_SAFETY_E2E)
642template <
typename M,
typename F,
typename>
644 const ::rclcpp::QoS& qos, F callback,
645 const ::rclcpp::SubscriptionOptions& options) {
652 out.user_fn_safety_ =
typename Subscription<M>::TypedSubscriptionSafetyFn(callback);
653 out.user_fn_ =
nullptr;
654 out.user_fn_ctx_ =
nullptr;
655 out.user_fn_info_ =
nullptr;
656 out.user_ctx_ =
nullptr;
658 uint8_t sched = (options.sched_context == ::rclcpp::SCHED_CONTEXT_UNSET)
660 : static_cast<uint8_t>(options.sched_context);
661 size_t handle =
static_cast<size_t>(-1);
663 nros_cpp_subscription_options_t ffi_options = nros_cpp_subscription_default_options();
664 ffi_options.sched_context = sched;
667 ffi_options.rx_buffer_hint =
static_cast<uint32_t
>(::rclcpp::rx_buffer_capacity<M>::value);
668 ffi_options.stamp_offset = ::rclcpp::stamp_offset<M>::value;
670 &handle_, topic, M::TYPE_NAME, M::TYPE_HASH, ffi_qos,
671 &Subscription<M>::message_safety_trampoline, &out, &
handle, &ffi_options);
673 out.sched_handle_id_ =
handle;
674 out.executor_ = executor_handle_;
677 out.store_topic_name(topic);
678 out.initialized_ =
true;
710template <typename F, typename Arg, typename =
void>
712template <
typename F,
typename Arg>
714 : ::
rclcpp::tr::true_type {};
729template <typename M, typename D, bool = sub_callable_with<D, const M&>::value>
731#if defined(NROS_CPP_HAS_SHARED_PTR)
732template <
typename M,
typename D,
bool = sub_callable_with<D, ::std::shared_ptr<const M>>::value>
733struct sub_shape_const_shared;
734template <
typename M,
typename D,
bool = sub_callable_with<D, ::std::shared_ptr<M>>::value>
735struct sub_shape_shared;
736template <
typename M,
typename D,
bool = sub_callable_with<D, ::std::unique_ptr<M>>::value>
737struct sub_shape_unique;
738template <
typename M,
typename D>
739struct sub_shape_unique<M, D, true> : ::rclcpp::tr::integral_constant<int, SUB_SHAPE_UNIQUE_PTR> {};
740template <
typename M,
typename D>
741struct sub_shape_unique<M, D, false> : ::rclcpp::tr::integral_constant<int, SUB_SHAPE_NONE> {};
742template <
typename M,
typename D>
743struct sub_shape_shared<M, D, true> : ::rclcpp::tr::integral_constant<int, SUB_SHAPE_SHARED_PTR> {};
744template <
typename M,
typename D>
struct sub_shape_shared<M, D, false> : sub_shape_unique<M, D> {};
745template <
typename M,
typename D>
746struct sub_shape_const_shared<M, D, true>
747 : ::rclcpp::tr::integral_constant<int, SUB_SHAPE_CONST_SHARED_PTR> {};
748template <
typename M,
typename D>
749struct sub_shape_const_shared<M, D, false> : sub_shape_shared<M, D> {};
750template <
typename M,
typename D>
751struct sub_shape_ref<M, D, false> : sub_shape_const_shared<M, D> {};
753template <
typename M,
typename D>
754struct sub_shape_ref<M, D, false> : ::rclcpp::tr::integral_constant<int, SUB_SHAPE_NONE> {};
756template <
typename M,
typename D>
757struct sub_shape_ref<M, D, true> : ::rclcpp::tr::integral_constant<int, SUB_SHAPE_REF> {};
760template <
typename M,
typename F>
770 template <
typename G>
static G&&
wrap(G&& f) {
return static_cast<G&&
>(f); }
780#if defined(NROS_CPP_HAS_SHARED_PTR)
784template <typename M, int Shape> struct subscription_heap_copy {
785 static ::std::shared_ptr<M> make(
const M& msg) { return ::std::make_shared<M>(msg); }
788 static ::std::unique_ptr<M> make(
const M& msg) { return ::std::unique_ptr<M>(
new M(msg)); }
791template <
typename M,
typename F,
int Shape>
struct subscription_callback_adapter {
793 typename ::rclcpp::tr::decay<F>::type fn;
794 void operator()(
const M& msg) { fn(subscription_heap_copy<M, Shape>::make(msg)); }
796 template <
typename G>
static Copying wrap(G&& f) {
return Copying{
static_cast<G&&
>(f)}; }
801template <
typename F,
typename A,
typename B,
typename =
void>
803template <
typename F,
typename A,
typename B>
806 : ::
rclcpp::tr::true_type {};
813template <
typename M,
typename D,
816#if defined(NROS_CPP_HAS_SHARED_PTR)
818 typename M,
typename D,
820struct sub_info_shape_const_shared;
821template <
typename M,
typename D,
823struct sub_info_shape_shared;
824template <
typename M,
typename D,
826struct sub_info_shape_unique;
827template <
typename M,
typename D>
828struct sub_info_shape_unique<M, D, true>
829 : ::rclcpp::tr::integral_constant<int, SUB_SHAPE_UNIQUE_PTR> {};
830template <
typename M,
typename D>
831struct sub_info_shape_unique<M, D, false> : ::rclcpp::tr::integral_constant<int, SUB_SHAPE_NONE> {};
832template <
typename M,
typename D>
833struct sub_info_shape_shared<M, D, true>
834 : ::rclcpp::tr::integral_constant<int, SUB_SHAPE_SHARED_PTR> {};
835template <
typename M,
typename D>
836struct sub_info_shape_shared<M, D, false> : sub_info_shape_unique<M, D> {};
837template <
typename M,
typename D>
838struct sub_info_shape_const_shared<M, D, true>
839 : ::rclcpp::tr::integral_constant<int, SUB_SHAPE_CONST_SHARED_PTR> {};
840template <
typename M,
typename D>
841struct sub_info_shape_const_shared<M, D, false> : sub_info_shape_shared<M, D> {};
842template <
typename M,
typename D>
843struct sub_info_shape_ref<M, D, false> : sub_info_shape_const_shared<M, D> {};
845template <
typename M,
typename D>
848template <
typename M,
typename D>
853template <
typename M,
typename F>
859 ::rclcpp::RmwMessageInfo r;
860 r.source_timestamp = c.source_timestamp;
861 r.received_timestamp = c.received_timestamp;
862 r.publication_sequence_number = c.publication_sequence_number;
863 r.reception_sequence_number = c.reception_sequence_number;
864 for (
int i = 0; i < 24; ++i)
865 r.publisher_gid.data[i] = c.publisher_gid[i];
867 return ::rclcpp::MessageInfo(r, c.has_info);
870#if defined(NROS_CPP_HAS_SHARED_PTR)
874template <typename M, typename F, int Shape> struct subscription_info_adapter {
876 typename ::rclcpp::tr::decay<F>::type fn;
877 void operator()(
const M& msg, const ::rclcpp::MessageInfo& info) {
878 fn(subscription_heap_copy<M, Shape>::make(msg), info);
881 template <
typename G>
static Copying wrap(G&& f) {
return Copying{
static_cast<G&&
>(f)}; }
883template <
typename M,
typename F>
struct subscription_info_adapter<M, F,
SUB_SHAPE_REF> {
884 template <
typename G>
static typename ::rclcpp::tr::decay<G>::type wrap(G&& f) {
885 return static_cast<G&&
>(f);
894template <
typename M,
typename Fn>
896 const QoS& qos, Fn fn,
size_t* out_handle_id) {
898 if (h ==
nullptr)
return Result(ErrorCode::NotInitialized);
899 Fn* heap =
new Fn(
static_cast<Fn&&
>(fn));
900 void (*invoke)(
const uint8_t*, size_t,
const nros_cpp_message_info_t*,
void*) =
901 [](
const uint8_t* data,
size_t len,
const nros_cpp_message_info_t* info,
void* ctx) {
903 if (M::ffi_deserialize(data, len, &msg) != 0 || info ==
nullptr)
return;
906 void (*drop)(
void*) = [](
void* ctx) {
delete static_cast<Fn*
>(ctx); };
908 nros_cpp_subscription_options_t opts = {};
909 opts.rx_buffer_hint =
static_cast<uint32_t
>(::rclcpp::rx_buffer_capacity<M>::value);
910 opts.stamp_offset = ::rclcpp::stamp_offset<M>::value;
911 return Result(nros_cpp_subscription_register_message_info(
912 h, topic, M::TYPE_NAME, M::TYPE_HASH, ffi_qos, invoke, heap, drop, out_handle_id, &opts));
927template <
typename M,
typename Fn>
929 const QoS& qos,
const Fn& fn,
size_t* out_handle_id) {
944 const Fn* self =
static_cast<const Fn*
>(ctx);
946 if (M::ffi_deserialize(data, len, &msg) != 0)
return;
953 void (*destroy)(
void*) = [](
void* ctx) {
static_cast<Fn*
>(ctx)->clear(); };
956 nros_cpp_subscription_options_t opts = {};
957 opts.rx_buffer_hint =
static_cast<uint32_t
>(::rclcpp::rx_buffer_capacity<M>::value);
958 opts.stamp_offset = ::rclcpp::stamp_offset<M>::value;
960 h, topic, M::TYPE_NAME, M::TYPE_HASH, ffi_qos, invoke,
961 reinterpret_cast<const uint8_t*
>(&fn),
sizeof(Fn), destroy, out_handle_id, &opts));
Definition executor.hpp:119
Result count_publishers(const char *topic_name, size_t *out_count)
Definition executor.hpp:411
Result create_subscription_in_group(const ::rclcpp::CallbackGroup &group, ::rclcpp::Subscription< M > &out, const char *topic, const ::rclcpp::QoS &qos, F callback, const ::rclcpp::SubscriptionOptions &options={})
const nros_cpp_node_t * ffi_handle() const
Definition node.hpp:1887
Result create_subscription_with_info(::rclcpp::Subscription< M > &out, const char *topic, const ::rclcpp::QoS &qos, F callback, const ::rclcpp::SubscriptionOptions &options={})
typename::rclcpp::Subscription< M >::SharedPtr create_subscription(const char *topic, const ::rclcpp::QoS &qos, Cb cb)
Definition subscription.hpp:197
bool has_sched_handle() const
Definition subscription.hpp:333
Result get_publisher_count(::rclcpp::Executor &executor, size_t *out_count) const
Definition subscription.hpp:261
Subscription(Subscription &&other)
Definition subscription.hpp:287
::rclcpp::SubscriptionHandle< M > ConstSharedPtr
Definition subscription.hpp:224
Subscription & operator=(Subscription &&other)
Definition subscription.hpp:301
size_t sched_handle_id() const
Definition subscription.hpp:334
Subscription()
Definition subscription.hpp:321
::rclcpp::SubscriptionHandle< M > UniquePtr
rclcpp::Subscription<M>::UniquePtr — see SharedPtr.
Definition subscription.hpp:226
::rclcpp::SubscriptionHandle< M > SharedPtr
Definition subscription.hpp:220
~Subscription()
Definition subscription.hpp:275
void(*)(const M &msg, void *ctx) TypedSubscriptionFnWithCtx
Definition subscription.hpp:232
const char * get_topic_name() const
Get the topic name.
Definition subscription.hpp:245
::rclcpp::QoS get_actual_qos() const
Definition subscription.hpp:358
void(*)(const M &msg, const uint8_t *attachment, size_t attachment_len) TypedSubscriptionInfoFn
Definition subscription.hpp:234
void(*)(const M &msg) TypedSubscriptionFn
Definition subscription.hpp:231
bool is_valid() const
Check if the subscription is initialized and valid.
Definition subscription.hpp:248
Inline storage-size macros for opaque entity buffers.
rclcpp::Executor — drives transport I/O and dispatches callbacks.
int nros_cpp_ret_t
Definition future.hpp:25
Definition rclcpp.hpp:418
SubscriptionCallbackShape
Definition subscription.hpp:718
@ SUB_SHAPE_CONST_SHARED_PTR
std::shared_ptr<const M> (M::ConstSharedPtr)
Definition subscription.hpp:721
@ SUB_SHAPE_REF
void(const M&) — the delivery path itself
Definition subscription.hpp:720
@ SUB_SHAPE_NONE
Definition subscription.hpp:719
@ SUB_SHAPE_SHARED_PTR
std::shared_ptr<M> (M::SharedPtr)
Definition subscription.hpp:722
@ SUB_SHAPE_UNIQUE_PTR
std::unique_ptr<M> (M::UniquePtr)
Definition subscription.hpp:723
constexpr nros_cpp_qos_t qos_to_ffi(const QoS &qos)
Definition qos.hpp:719
T && sub_declval() noexcept
Unevaluated-only declval, so the detection below needs no <utility>.
Result register_subscription_capturing(::rclcpp::Node &node, const char *topic, const QoS &qos, const Fn &fn, size_t *out_handle_id)
Definition subscription.hpp:928
inline ::rclcpp::MessageInfo message_info_from_ffi(const nros_cpp_message_info_t &c)
Issue 2036 — the MessageInfo the runtime reported, in upstream's layout.
Definition subscription.hpp:858
::nros_cpp_integrity_status_t IntegrityStatus
Definition subscription.hpp:135
@ NotInitialized
Definition result.hpp:131
static constexpr size_t SUBSCRIPTION_TOPIC_NAME_MAX
Definition subscription.hpp:143
ResultOf< void > Result
Definition result.hpp:247
rclcpp::Node and global session helpers.
rclcpp::QoS — full DDS-shaped QoS settings (Phase 108.B.7).
**auto handle
Definition client.hpp:127
rclcpp::Result, rclcpp::ErrorCode, and the RCLCPP_TRY macro.
rclcpp::Stream<T> — multi-shot message receiver.
Issue 2036 — true iff F& can be called with (A, B).
Definition subscription.hpp:802
true iff F& can be called with one argument of type Arg.
Definition subscription.hpp:711
Definition subscription.hpp:815
Definition subscription.hpp:730
Definition subscription.hpp:774
void operator()(const M &)
Definition subscription.hpp:775
static Refused wrap(G &&)
Definition subscription.hpp:777
static G && wrap(G &&f)
Definition subscription.hpp:770
Definition subscription.hpp:767
Which SubscriptionCallbackShape the callable F is, for message M.
Definition subscription.hpp:761
Definition subscription.hpp:855
nros_cpp_ret_t nros_cpp_subscription_register_capturing(const nros_cpp_node_t *node, const char *topic, const char *type_name, const char *type_hash, nros_cpp_qos_t qos, nros_cpp_subscription_message_callback_t callback, const uint8_t *capture, size_t capture_len, void(*capture_drop)(void *capture), size_t *out_handle_id, const nros_cpp_subscription_options_t *options)
nros_cpp_ret_t nros_cpp_subscription_register(const nros_cpp_node_t *node, const char *topic, const char *type_name, const char *type_hash, nros_cpp_qos_t qos, nros_cpp_subscription_message_callback_t callback, void *context, size_t *out_handle_id, const nros_cpp_subscription_options_t *options)
void(* nros_cpp_subscription_message_info_callback_t)(const uint8_t *data, size_t len, const uint8_t *attachment, size_t attachment_len, void *ctx)
Definition subscription.hpp:100
nros_cpp_ret_t nros_cpp_subscription_register_with_info(const nros_cpp_node_t *node, const char *topic, const char *type_name, const char *type_hash, nros_cpp_qos_t qos, nros_cpp_subscription_message_info_callback_t callback, void *context, size_t *out_handle_id, const nros_cpp_subscription_options_t *options)
void(* nros_cpp_subscription_message_callback_t)(const uint8_t *data, size_t len, void *ctx)
Definition subscription.hpp:57