nros C++ API
Lightweight ROS 2 client for embedded real-time systems (C++ headers)
Loading...
Searching...
No Matches
subscription.hpp
Go to the documentation of this file.
1// nros-cpp: Subscription class
2// Freestanding C++ — no exceptions, no STL required
3
13
14#ifndef NROS_RCLCPP_SUBSCRIPTION_HPP
15#define NROS_RCLCPP_SUBSCRIPTION_HPP
16
17#include <cstdint>
18#include <cstddef>
19#include <string.h> // memcpy — `<cstring>` isn't in Zephyr's minimal libcpp
20
21#include "rclcpp/detail/traits.hpp"
23#include "rclcpp/detail/entity_name.hpp" // phase-444 — the one entity-name copy
24#include "rclcpp/executor.hpp" // phase-444 — the graph counter these forward to
25#include "rclcpp/result.hpp"
26#include "rclcpp/subscription_handle.hpp"
27#include "rclcpp/inplace_fn.hpp"
28#include "rclcpp/size_bound.hpp" // rclcpp::rx_buffer_capacity<M> — the receive-buffer size
29// RFC-0088 D5 — NROS_CPP_ASSERT_MESSAGE_FORMAT, expanded in the creators below.
31#include "rclcpp/stream.hpp"
32
33// phase-417 W1.a — `<memory>` for the nested pointer aliases below.
34// `NROS_CPP_HAS_SHARED_PTR` and the other five capability macros have ONE
35// definition site, and the measured reason the predicate needs both probes
36// (issues 0112, 1187, 1240) is stated there.
37#include "rclcpp/detail/std_detect.hpp"
38
39#include "rclcpp/detail/nros_cpp_ffi.h"
40
41// issue 1437 — `get_actual_qos()` returns a `rclcpp::QoS` BY VALUE from an
42// inline body, so the complete type must be here, not only by the time
43// `rclcpp/node.hpp` is pulled in below.
44//
45// AFTER `nros_cpp_ffi.h`, never before: `qos.hpp` defines the four
46// `nros_cpp_qos_*_t` enums ITSELF under `#ifndef NROS_CPP_FFI_H`, so
47// reaching it first makes the cbindgen header a REDEFINITION of all four.
48#include "rclcpp/qos.hpp"
49#include "rclcpp/message_info.hpp" // issue 2036 — the two-argument callback shapes
50
51// Phase 189.M3.x — `nros_cpp_subscription_register` is excluded from cbindgen
52// (its Rust signature uses `RawSubscriptionCallback`, an external-crate type
53// alias cbindgen names without defining). Declare it locally with a plain
54// function-pointer typedef matching the ABI (`void(data, len, ctx)`), mirroring
55// the service.hpp callback-register treatment.
56extern "C" {
57typedef void (*nros_cpp_subscription_message_callback_t)(const uint8_t* data, size_t len,
58 void* ctx);
59
60// phase-402: the trailing `sched_context` / `callback_group` arguments moved
61// into `nros_cpp_subscription_options_t` (defined by the cbindgen output above);
62// a NULL `options` is all-defaults, i.e. the pre-phase-402 behaviour.
63nros_cpp_ret_t nros_cpp_subscription_register(const nros_cpp_node_t* node, const char* topic,
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);
69
93 const nros_cpp_node_t* node, const char* topic, const char* type_name, const char* type_hash,
94 nros_cpp_qos_t qos, nros_cpp_subscription_message_callback_t callback, const uint8_t* capture,
95 size_t capture_len, void (*capture_drop)(void* capture), size_t* out_handle_id,
96 const nros_cpp_subscription_options_t* options);
97
98// Phase 189.M3.4 — callback-style register that also delivers the sample's wire
99// attachment (5-arg trampoline). Same cbindgen-exclusion reason as above.
100typedef void (*nros_cpp_subscription_message_info_callback_t)(const uint8_t* data, size_t len,
101 const uint8_t* attachment,
102 size_t attachment_len, void* ctx);
103
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);
108// Phase 269 W3 — callback-style subscription that delivers the sample's E2E
109// integrity status alongside the CDR bytes. Same cbindgen-exclusion reason as
110// above (takes `RawSubscriptionSafetyCallback`, an external nros-node alias
111// gated on the `safety-e2e` feature). The 6-arg trampoline unpacks the three
112// integrity scalars; `subscription.hpp` repacks them into
113// `nros_cpp_integrity_status_t` for the typed user handler. Gated on the
114// `NROS_SAFETY_E2E` build feature (lowered from `[system].features =
115// ["safety"]` via `NanoRosCapabilities.cmake`).
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);
120
121nros_cpp_ret_t nros_cpp_subscription_register_validated(
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);
125#endif // NROS_SAFETY_E2E
126} // extern "C"
127
128namespace rclcpp {
129
135using IntegrityStatus = ::nros_cpp_integrity_status_t;
136
143static constexpr size_t SUBSCRIPTION_TOPIC_NAME_MAX = 256;
144
145} // namespace rclcpp
146
154namespace rclcpp {
155class Node;
156}
157
158// ============================================================================
159// `rclcpp::Subscription<M>` -- DEFINED here (RFC-0089: rclcpp:: is the home)
160// ============================================================================
161//
162// phase-428: the definition moved from `nros::` to `rclcpp::` (the `nros::`
163// alias went in phase-483 W1). The nested `SharedPtr` / `ConstSharedPtr` / `UniquePtr`
164// aliases live on the class itself, so `rclcpp::Subscription<M>::SharedPtr`
165// resolves with no wrapper type in between.
166namespace rclcpp {
167
197template <typename M> class Subscription {
198 public:
227
231 using TypedSubscriptionFn = void (*)(const M& msg);
232 using TypedSubscriptionFnWithCtx = void (*)(const M& msg, void* ctx);
233 // Phase 189.M3.4 — callback-with-attachment handler (bridge_origin etc.).
234 using TypedSubscriptionInfoFn = void (*)(const M& msg, const uint8_t* attachment,
235 size_t attachment_len);
236 // Phase 269 W3 — callback-with-integrity handler: receives the deserialized
237 // message plus the sample's E2E CRC/sequence status. Requires
238 // `NROS_SAFETY_E2E` (lowered from `[system].features = ["safety"]`).
239#if defined(NROS_SAFETY_E2E)
240 using TypedSubscriptionSafetyFn = void (*)(const M& msg,
241 const ::rclcpp::IntegrityStatus& integrity);
242#endif // NROS_SAFETY_E2E
243
245 const char* get_topic_name() const { return initialized_ ? topic_name_ : ""; }
246
248 bool is_valid() const { return initialized_; }
249
261 Result get_publisher_count(::rclcpp::Executor& executor, size_t* out_count) const {
262 if (!initialized_) return Result(::rclcpp::ErrorCode::NotInitialized);
263 return executor.count_publishers(topic_name_, out_count);
264 }
265
275 ~Subscription() { initialized_ = false; }
276
277 // Move semantics (non-copyable). Bookkeeping only — there is no storage to
278 // relocate, so `nros_cpp_subscription_relocate` is not called here.
279 //
280 // THE HAZARD, narrowed to its real subject: the out-ref creators below hand
281 // the arena `&out` as the trampoline context, so an object registered that
282 // way must NOT be moved afterwards — the move copies the handler pointers
283 // and leaves the arena dispatching into the source object's address. The
284 // returning `create_subscription` (`nros.hpp`) has no such hazard: W1 put
285 // the capture in the arena and it hands back a `SharedPtr`, which is a
286 // value with nothing pointing at it.
287 Subscription(Subscription&& other) : initialized_(other.initialized_) {
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_;
295#endif // NROS_SAFETY_E2E
296 sched_handle_id_ = other.sched_handle_id_;
297 executor_ = other.executor_;
298 other.initialized_ = false;
299 }
300
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_;
311#endif // NROS_SAFETY_E2E
312 sched_handle_id_ = other.sched_handle_id_;
313 executor_ = other.executor_;
314 other.initialized_ = false;
315 }
316 return *this;
317 }
318
321 Subscription() : topic_name_{}, initialized_(false) {}
322
333 bool has_sched_handle() const { return sched_handle_id_ != static_cast<size_t>(-1); }
334 size_t sched_handle_id() const { return sched_handle_id_; }
335
359 nros_cpp_qos_t f{};
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();
363 }
364 return ::rclcpp::detail::qos_from_ffi(f);
365 }
366
367 private:
368 Subscription(const Subscription&) = delete;
369 Subscription& operator=(const Subscription&) = delete;
370
371 friend class ::rclcpp::Node;
372
376 static void message_trampoline(const uint8_t* data, size_t len, void* ctx) {
377 auto* self = static_cast<Subscription*>(ctx);
378 if (self == nullptr) return;
379 M msg;
380 if (M::ffi_deserialize(data, len, &msg) != 0) return;
381 if (self->user_fn_ != nullptr) {
382 self->user_fn_(msg);
383 } else if (self->user_fn_ctx_ != nullptr) {
384 self->user_fn_ctx_(msg, self->user_ctx_);
385 }
386 }
387
391 static void message_info_trampoline(const uint8_t* data, size_t len, const uint8_t* attachment,
392 size_t attachment_len, void* ctx) {
393 auto* self = static_cast<Subscription*>(ctx);
394 if (self == nullptr) return;
395 M msg;
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);
399 }
400 }
401
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) {
410 auto* self = static_cast<Subscription*>(ctx);
411 if (self == nullptr) return;
412 M msg;
413 if (M::ffi_deserialize(data, len, &msg) != 0) return;
414 if (self->user_fn_safety_ != nullptr) {
415 nros_cpp_integrity_status_t status;
416 status.gap = gap;
417 status.duplicate = duplicate;
418 status.crc_valid = crc_valid;
419 self->user_fn_safety_(msg, status);
420 }
421 }
422#endif // NROS_SAFETY_E2E
423
430 void store_topic_name(const char* topic) {
431 size_t n = 0;
432 while (topic[n] != '\0' && n + 1 < sizeof(topic_name_)) {
433 topic_name_[n] = topic[n];
434 ++n;
435 }
436 topic_name_[n] = '\0';
437 }
438
439 // phase-456 W2b — `storage_` (NROS_SUBSCRIBER_SIZE bytes) and `stream_` are
440 // GONE, with the taking API that was their only reader. On this path the
441 // arena owns the subscriber, so those bytes were never filled, and
442 // `take()` reinterpreted them (`&mut *(storage as *mut RmwSubscriber)`)
443 // rather than refusing. ~~`callback_mode_`~~ went with them: it distinguished
444 // the two owners inside one class, and there is one owner here now.
446 bool initialized_;
447 // Phase 189.M3.1 — executor HandleId for sched-context binding, or
448 // SIZE_MAX (the default) when the registration has not happened yet.
449 size_t sched_handle_id_ = static_cast<size_t>(-1);
450 // The executor that owns the arena entry `sched_handle_id_` indexes —
451 // the phase-467 RMW gap-closure design study's Row 13.
452 //
453 // A `HandleId` is EXECUTOR-SCOPED: an index into one executor's arena, not
454 // a process-wide identity, and RFC-0002 puts one executor on one RTOS
455 // task, so a tiered image has several. That is why `get_actual_qos()`
456 // could not be answered from `sched_handle_id_` alone, and why resolving
457 // it "from the sched handle" would have needed a global executor
458 // registry. The pair is the same one `rclcpp::Client` already keeps
459 // (`{executor_, handle_id_}`), for the same reason.
460 void* executor_ = nullptr;
461 // The user's handler, dispatched by `message_trampoline` during spin.
462 TypedSubscriptionFn user_fn_ = nullptr;
463 TypedSubscriptionFnWithCtx user_fn_ctx_ = nullptr;
464 TypedSubscriptionInfoFn user_fn_info_ = nullptr;
465 void* user_ctx_ = nullptr;
466#if defined(NROS_SAFETY_E2E)
467 // Phase 269 W3 — handler for the integrity-carrying callback path; nullptr
468 // when not using `create_subscription_with_safety`.
469 TypedSubscriptionSafetyFn user_fn_safety_ = nullptr;
470#endif // NROS_SAFETY_E2E
471};
472
473} // namespace rclcpp
474
475// Phase 84.G8: out-of-line definition of Node::create_subscription<M>().
476#include "rclcpp/node.hpp"
477
478namespace rclcpp {
479
480// Phase 189.M3.x — callback-style (arena-registered) subscription. The arena
481// owns the subscriber + dispatches `out`'s message handler during spin_once, so
482// the handle is real and `options.sched_context` is functional. Mirrors the
483// callback-style `create_service` one entity over.
484} // namespace rclcpp
485
486namespace rclcpp {
487template <typename M, typename F, typename>
488Result Node::create_subscription(Subscription<M>& out, const char* topic, const ::rclcpp::QoS& qos,
489 F callback, const ::rclcpp::SubscriptionOptions& options) {
490 // RFC-0088 D5 — one image, one backend, one encoding. Compile-time, so a
491 // message the linked backend cannot encode never reaches the wire.
493 if (!initialized_) return Result(::rclcpp::ErrorCode::NotInitialized);
495
496 // Store the user handler (compile error if F isn't convertible to the
497 // plain-fn-ptr handler type).
498 out.user_fn_ = typename Subscription<M>::TypedSubscriptionFn(callback);
499 out.user_fn_ctx_ = nullptr;
500 out.user_ctx_ = nullptr;
501
502 uint8_t sched = (options.sched_context == ::rclcpp::SCHED_CONTEXT_UNSET)
503 ? 0u
504 : static_cast<uint8_t>(options.sched_context);
505 size_t handle = static_cast<size_t>(-1);
506 // phase-402: `sched_context` is a FIELD now. `callback_group` stays unset,
507 // i.e. the default group.
508 nros_cpp_subscription_options_t ffi_options = nros_cpp_subscription_default_options();
509 ffi_options.sched_context = sched;
510 // phase-456 W7 — state the bound. `M` is a template parameter of this
511 // function, so the number was always available here; leaving it at the
512 // option default put this registration on issue 1319's ~~`c_raw_no_hint`~~ row
513 // while the sizing descriptor credited the entry with a supplied hint.
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; // issue 2215
517 &handle_, topic, M::TYPE_NAME, M::TYPE_HASH, ffi_qos, &Subscription<M>::message_trampoline,
518 &out, &handle, &ffi_options);
519 if (ret == 0) {
520 out.sched_handle_id_ = handle;
521 out.executor_ = executor_handle_;
522 // phase-456 W2b — `get_topic_name()` answered "" on every callback-style
523 // subscription until this line existed.
524 out.store_topic_name(topic);
525 out.initialized_ = true;
526 }
527 return Result(ret);
528}
529} // namespace rclcpp
530
531namespace rclcpp {
532
533// Phase 273 (RFC-0047) — callback-style subscription **in** a named callback group.
534// Mirrors create_subscription (callback-style) exactly but passes group.get_name()
535// as `callback_group` so the executor binds the slot via group_sched_table.
536} // namespace rclcpp
537
538namespace rclcpp {
539template <typename M, typename F, typename>
540Result Node::create_subscription_in_group(const ::rclcpp::CallbackGroup& group,
541 Subscription<M>& out, const char* topic,
542 const ::rclcpp::QoS& qos, F callback,
543 const ::rclcpp::SubscriptionOptions& options) {
544 // RFC-0088 D5 — one image, one backend, one encoding. Compile-time, so a
545 // message the linked backend cannot encode never reaches the wire.
547 if (!initialized_) return Result(::rclcpp::ErrorCode::NotInitialized);
549
550 out.user_fn_ = typename Subscription<M>::TypedSubscriptionFn(callback);
551 out.user_fn_ctx_ = nullptr;
552 out.user_ctx_ = nullptr;
553
554 uint8_t sched = (options.sched_context == ::rclcpp::SCHED_CONTEXT_UNSET)
555 ? 0u
556 : static_cast<uint8_t>(options.sched_context);
557 size_t handle = static_cast<size_t>(-1);
558 // phase-402: both the sched slot and the Phase 273 group name are FIELDS now.
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();
562 // phase-456 W7 — see the ungrouped form above. A grouped subscription costs
563 // the arena exactly what an ungrouped one does, so it states the same bound.
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; // issue 2215
567 &handle_, topic, M::TYPE_NAME, M::TYPE_HASH, ffi_qos, &Subscription<M>::message_trampoline,
568 &out, &handle, &ffi_options);
569 if (ret == 0) {
570 out.sched_handle_id_ = handle;
571 out.executor_ = executor_handle_;
572 // phase-456 W2b — `get_topic_name()` answered "" on every callback-style
573 // subscription until this line existed.
574 out.store_topic_name(topic);
575 out.initialized_ = true;
576 }
577 return Result(ret);
578}
579} // namespace rclcpp
580
581namespace rclcpp {
582
583// Phase 189.M3.4 — callback-style subscription that delivers the wire attachment.
584// Mirrors the callback `create_subscription` one step over, but stores the
585// `(const M&, attachment, att_len)` handler + registers via the with-info arena
586// path so the trampoline receives the attachment.
587} // namespace rclcpp
588
589namespace rclcpp {
590template <typename M, typename F, typename>
592 const ::rclcpp::QoS& qos, F callback,
593 const ::rclcpp::SubscriptionOptions& options) {
594 // RFC-0088 D5 — one image, one backend, one encoding. Compile-time, so a
595 // message the linked backend cannot encode never reaches the wire.
597 if (!initialized_) return Result(::rclcpp::ErrorCode::NotInitialized);
599
600 out.user_fn_info_ = typename Subscription<M>::TypedSubscriptionInfoFn(callback);
601 out.user_fn_ = nullptr;
602 out.user_fn_ctx_ = nullptr;
603 out.user_ctx_ = nullptr;
604
605 uint8_t sched = (options.sched_context == ::rclcpp::SCHED_CONTEXT_UNSET)
606 ? 0u
607 : static_cast<uint8_t>(options.sched_context);
608 size_t handle = static_cast<size_t>(-1);
609 // phase-402: `sched_context` is a FIELD now.
610 nros_cpp_subscription_options_t ffi_options = nros_cpp_subscription_default_options();
611 ffi_options.sched_context = sched;
612 // phase-456 W7 — the attachment rides beside the sample and does not change
613 // how many bytes the sample itself needs, so this is the same bound the
614 // plain callback form states.
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; // issue 2215
618 &handle_, topic, M::TYPE_NAME, M::TYPE_HASH, ffi_qos,
619 &Subscription<M>::message_info_trampoline, &out, &handle, &ffi_options);
620 if (ret == 0) {
621 out.sched_handle_id_ = handle;
622 out.executor_ = executor_handle_;
623 // phase-456 W2b — `get_topic_name()` answered "" on every callback-style
624 // subscription until this line existed.
625 out.store_topic_name(topic);
626 out.initialized_ = true;
627 }
628 return Result(ret);
629}
630} // namespace rclcpp
631
632namespace rclcpp {
633
634#if defined(NROS_SAFETY_E2E)
635// Phase 269 W3 — out-of-line definition of Node::create_subscription_with_safety.
636// Mirrors `create_subscription_with_info` one overload over, but routes through
637// `nros_cpp_subscription_register_validated` so the arena dispatches the
638// `message_safety_trampoline` on each new sample.
639} // namespace rclcpp
640
641namespace rclcpp {
642template <typename M, typename F, typename>
643Result Node::create_subscription_with_safety(Subscription<M>& out, const char* topic,
644 const ::rclcpp::QoS& qos, F callback,
645 const ::rclcpp::SubscriptionOptions& options) {
646 // RFC-0088 D5 — one image, one backend, one encoding. Compile-time, so a
647 // message the linked backend cannot encode never reaches the wire.
649 if (!initialized_) return Result(::rclcpp::ErrorCode::NotInitialized);
650 nros_cpp_qos_t ffi_qos = ::rclcpp::detail::qos_to_ffi(qos);
651
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;
657
658 uint8_t sched = (options.sched_context == ::rclcpp::SCHED_CONTEXT_UNSET)
659 ? 0u
660 : static_cast<uint8_t>(options.sched_context);
661 size_t handle = static_cast<size_t>(-1);
662 // phase-402: `sched_context` is a FIELD now.
663 nros_cpp_subscription_options_t ffi_options = nros_cpp_subscription_default_options();
664 ffi_options.sched_context = sched;
665 // phase-456 W7 — the validated path deserializes the same bytes; the safety
666 // status is computed from them, not received alongside them.
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; // issue 2215
669 nros_cpp_ret_t ret = nros_cpp_subscription_register_validated(
670 &handle_, topic, M::TYPE_NAME, M::TYPE_HASH, ffi_qos,
671 &Subscription<M>::message_safety_trampoline, &out, &handle, &ffi_options);
672 if (ret == 0) {
673 out.sched_handle_id_ = handle;
674 out.executor_ = executor_handle_;
675 // phase-456 W2b — `get_topic_name()` answered "" on every callback-style
676 // subscription until this line existed.
677 out.store_topic_name(topic);
678 out.initialized_ = true;
679 }
680 return Result(ret);
681}
682} // namespace rclcpp
683
684namespace rclcpp {
685#endif // NROS_SAFETY_E2E
686
687namespace detail {
688
689// ---- issue 1995: which callback shapes a subscription takes ------------------
690//
691// The arena delivers `const M&` from a message decoded in its dispatch frame,
692// and that stays the ONE delivery path. Upstream's other common shapes —
693// `M::UniquePtr`, `M::SharedPtr`, `M::ConstSharedPtr` (by value or `const&`),
694// `std::shared_ptr<const M>` — are taken on a HOSTED build by wrapping the
695// callable in an adapter that COPIES the message into a heap object per
696// delivery and hands the callback that pointer.
697//
698// Why a copy, not the service handler's frame loan (`detail::lend_from_frame`):
699// a subscription callback KEEPING its pointer (`last_msg_ = msg;`) is the
700// common case, not the rare one, and a loan turns it into an abort. A copy
701// makes keeping it legal, and the allocation is opted into by the signature —
702// a `const M&` callback still allocates nothing. Freestanding builds have no
703// allocator on this path, so there the pointer shapes are refused at compile
704// time (`NROS_RCLCPP_REFUSE_SUBSCRIPTION_CALLBACK_SHAPE`).
705
707template <typename T> T&& sub_declval() noexcept;
708
710template <typename F, typename Arg, typename = void>
711struct sub_callable_with : ::rclcpp::tr::false_type {};
712template <typename F, typename Arg>
713struct sub_callable_with<F, Arg, decltype((void)sub_declval<F&>()(sub_declval<Arg>()))>
714 : ::rclcpp::tr::true_type {};
715
725
726// Each step instantiates its probe only when every earlier one failed, so a
727// generic lambda that IS callable with `const M&` is never instantiated with a
728// smart pointer (its body would be a hard error, not a substitution failure).
729template <typename M, typename D, bool = sub_callable_with<D, const M&>::value>
731#if defined(NROS_CPP_HAS_SHARED_PTR) // hosted-family: shared-ptr-interop
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> {};
752#else
753template <typename M, typename D>
754struct sub_shape_ref<M, D, false> : ::rclcpp::tr::integral_constant<int, SUB_SHAPE_NONE> {};
755#endif // NROS_CPP_HAS_SHARED_PTR
756template <typename M, typename D>
757struct sub_shape_ref<M, D, true> : ::rclcpp::tr::integral_constant<int, SUB_SHAPE_REF> {};
758
760template <typename M, typename F>
761struct subscription_callback_shape : sub_shape_ref<M, typename ::rclcpp::tr::decay<F>::type> {};
762
767template <typename M, typename F, int Shape> struct subscription_callback_adapter;
768
769template <typename M, typename F> struct subscription_callback_adapter<M, F, SUB_SHAPE_REF> {
770 template <typename G> static G&& wrap(G&& f) { return static_cast<G&&>(f); }
771};
772
773template <typename M, typename F> struct subscription_callback_adapter<M, F, SUB_SHAPE_NONE> {
774 struct Refused {
775 void operator()(const M&) {}
776 };
777 template <typename G> static Refused wrap(G&&) { return Refused(); }
778};
779
780#if defined(NROS_CPP_HAS_SHARED_PTR) // hosted-family: shared-ptr-interop
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); }
786};
787template <typename M> struct subscription_heap_copy<M, SUB_SHAPE_UNIQUE_PTR> {
788 static ::std::unique_ptr<M> make(const M& msg) { return ::std::unique_ptr<M>(new M(msg)); }
789};
790
791template <typename M, typename F, int Shape> struct subscription_callback_adapter {
792 struct Copying {
793 typename ::rclcpp::tr::decay<F>::type fn;
794 void operator()(const M& msg) { fn(subscription_heap_copy<M, Shape>::make(msg)); }
795 };
796 template <typename G> static Copying wrap(G&& f) { return Copying{static_cast<G&&>(f)}; }
797};
798#endif // NROS_CPP_HAS_SHARED_PTR
799
801template <typename F, typename A, typename B, typename = void>
802struct sub_callable_with2 : ::rclcpp::tr::false_type {};
803template <typename F, typename A, typename B>
804struct sub_callable_with2<F, A, B,
805 decltype((void)sub_declval<F&>()(sub_declval<A>(), sub_declval<B>()))>
806 : ::rclcpp::tr::true_type {};
807
813template <typename M, typename D,
816#if defined(NROS_CPP_HAS_SHARED_PTR) // hosted-family: shared-ptr-interop
817template <
818 typename M, typename D,
819 bool = sub_callable_with2<D, ::std::shared_ptr<const M>, const ::rclcpp::MessageInfo&>::value>
820struct sub_info_shape_const_shared;
821template <typename M, typename D,
822 bool = sub_callable_with2<D, ::std::shared_ptr<M>, const ::rclcpp::MessageInfo&>::value>
823struct sub_info_shape_shared;
824template <typename M, typename D,
825 bool = sub_callable_with2<D, ::std::unique_ptr<M>, const ::rclcpp::MessageInfo&>::value>
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> {};
844#else
845template <typename M, typename D>
846struct sub_info_shape_ref<M, D, false> : ::rclcpp::tr::integral_constant<int, SUB_SHAPE_NONE> {};
847#endif // NROS_CPP_HAS_SHARED_PTR
848template <typename M, typename D>
849struct sub_info_shape_ref<M, D, true> : ::rclcpp::tr::integral_constant<int, SUB_SHAPE_REF> {};
850
853template <typename M, typename F>
855 : sub_info_shape_ref<M, typename ::rclcpp::tr::decay<F>::type> {};
856
858inline ::rclcpp::MessageInfo message_info_from_ffi(const nros_cpp_message_info_t& c) {
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];
866 // Issue 2074 — carry "the backend reported nothing" through.
867 return ::rclcpp::MessageInfo(r, c.has_info);
868}
869
870#if defined(NROS_CPP_HAS_SHARED_PTR) // hosted-family: shared-ptr-interop
874template <typename M, typename F, int Shape> struct subscription_info_adapter {
875 struct Copying {
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);
879 }
880 };
881 template <typename G> static Copying wrap(G&& f) { return Copying{static_cast<G&&>(f)}; }
882};
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);
886 }
887};
888
894template <typename M, typename Fn>
895inline Result register_subscription_message_info(::rclcpp::Node& node, const char* topic,
896 const QoS& qos, Fn fn, size_t* out_handle_id) {
897 const nros_cpp_node_t* h = node.ffi_handle();
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) {
902 M msg;
903 if (M::ffi_deserialize(data, len, &msg) != 0 || info == nullptr) return;
904 (*static_cast<Fn*>(ctx))(static_cast<const M&>(msg), message_info_from_ffi(*info));
905 };
906 void (*drop)(void*) = [](void* ctx) { delete static_cast<Fn*>(ctx); };
907 nros_cpp_qos_t ffi_qos = detail::qos_to_ffi(qos);
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; // issue 2215
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));
913}
914#endif // NROS_CPP_HAS_SHARED_PTR
915
927template <typename M, typename Fn>
929 const QoS& qos, const Fn& fn, size_t* out_handle_id) {
930 // phase-456 W8 — the per-entry capture budget this used to assert is GONE.
931 // It read `sizeof(Fn) <= NROS_CPP_CALLBACK_CAPACITY + 2 * sizeof(void*)` and
932 // named ~~`CALLBACK_CAPTURE_BYTES`~~, an arena constant that no longer exists:
933 // the runtime allocates exactly `capture_len` bytes now, so there is no
934 // cross-language number for the two sides to agree about and nothing here
935 // to assert. `InplaceFn`'s own capacity `static_assert` is untouched and is
936 // still the knob a too-large CALLABLE hits (`inplace_fn.hpp`), which is a
937 // property of that type rather than of this registration.
938 const nros_cpp_node_t* h = node.ffi_handle();
939 if (h == nullptr) return Result(ErrorCode::NotInitialized);
940
941 nros_cpp_subscription_message_callback_t invoke = [](const uint8_t* data, size_t len,
942 void* ctx) {
943 // `ctx` is the ARENA's copy of `fn`, not ours.
944 const Fn* self = static_cast<const Fn*>(ctx);
945 M msg;
946 if (M::ffi_deserialize(data, len, &msg) != 0) return;
947 (*self)(msg);
948 };
949
950 // phase-476 W2 — the arena's copy is destroyed when the subscription is
951 // released. `Fn` is an `rclcpp::InplaceFn` (the one caller passes one), so
952 // `clear()` runs the stored callable's own destructor.
953 void (*destroy)(void*) = [](void* ctx) { static_cast<Fn*>(ctx)->clear(); };
954
955 nros_cpp_qos_t ffi_qos = detail::qos_to_ffi(qos);
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; // issue 2215
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));
962}
963
964} // namespace detail
965
966} // namespace rclcpp
967
968// ============================================================================
969// WHAT LEFT THIS HEADER
970// ============================================================================
971//
972// phase-456 W2 deleted `rclcpp::detail::SubscriptionCallback<M>` — the heap
973// cell whose only job was to give a capturing lambda a stable address. W1 put
974// the capture in the arena entry, so the cell had nothing left to hold, and
975// `nros.hpp`'s returning `create_subscription` stopped allocating one.
976//
977// phase-456 W2b moved the TAKING API — `take`, `take_sized`, `take_serialized`,
978// `take_serialized_with_attachment`, `take_validated`, `take_validated_sized`,
979// `take_sequence`, `try_borrow` + `View`, `stream()` and the three QoS-event setters — to
980// `rclcpp::PollSubscription<M>` in `rclcpp/polling_subscription.hpp`, together with
981// the out-ref creator that fills it. Those calls all dereference `storage_`,
982// which only a caller-owned subscriber ever has.
983
984#endif // NROS_RCLCPP_SUBSCRIPTION_HPP
Definition executor.hpp:119
Result count_publishers(const char *topic_name, size_t *out_count)
Definition executor.hpp:411
Definition node.hpp:845
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 qos.hpp:276
Definition node.hpp:200
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
Definition rclcpp.hpp:75
::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::format_of<M> — a message type's serialization format; rclcpp::linked_format() — the linked ba...
#define NROS_CPP_ASSERT_MESSAGE_FORMAT(M)
Definition serialization_format.hpp:97
rclcpp::Stream<T> — multi-shot message receiver.
Definition qos.hpp:84
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
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
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