subscription.hpp

Go to the documentation of this file.

// nros-cpp: Subscription class
// Freestanding C++ — no exceptions, no STL required
 
#ifndef NROS_RCLCPP_SUBSCRIPTION_HPP
#define NROS_RCLCPP_SUBSCRIPTION_HPP
 
#include <cstdint>
#include <cstddef>
#include <string.h> // memcpy — `<cstring>` isn't in Zephyr's minimal libcpp
 
#include "rclcpp/detail/traits.hpp"
#include "rclcpp/detail/config.hpp"
#include "rclcpp/detail/entity_name.hpp" // phase-444 — the one entity-name copy
#include "rclcpp/executor.hpp"           // phase-444 — the graph counter these forward to
#include "rclcpp/result.hpp"
#include "rclcpp/subscription_handle.hpp"
#include "rclcpp/inplace_fn.hpp"
#include "rclcpp/size_bound.hpp" // rclcpp::rx_buffer_capacity<M> — the receive-buffer size
// RFC-0088 D5 — NROS_CPP_ASSERT_MESSAGE_FORMAT, expanded in the creators below.
#include "rclcpp/detail/serialization_format.hpp"
#include "rclcpp/stream.hpp"
 
// phase-417 W1.a — `<memory>` for the nested pointer aliases below.
// `NROS_CPP_HAS_SHARED_PTR` and the other five capability macros have ONE
// definition site, and the measured reason the predicate needs both probes
// (issues 0112, 1187, 1240) is stated there.
#include "rclcpp/detail/std_detect.hpp"
 
#include "rclcpp/detail/nros_cpp_ffi.h"
 
// issue 1437 — `get_actual_qos()` returns a `rclcpp::QoS` BY VALUE from an
// inline body, so the complete type must be here, not only by the time
// `rclcpp/node.hpp` is pulled in below.
//
// AFTER `nros_cpp_ffi.h`, never before: `qos.hpp` defines the four
// `nros_cpp_qos_*_t` enums ITSELF under `#ifndef NROS_CPP_FFI_H`, so
// reaching it first makes the cbindgen header a REDEFINITION of all four.
#include "rclcpp/qos.hpp"
 
// Phase 189.M3.x — `nros_cpp_subscription_register` is excluded from cbindgen
// (its Rust signature uses `RawSubscriptionCallback`, an external-crate type
// alias cbindgen names without defining). Declare it locally with a plain
// function-pointer typedef matching the ABI (`void(data, len, ctx)`), mirroring
// the service.hpp callback-register treatment.
extern "C" {
typedef void (*nros_cpp_subscription_message_callback_t)(const uint8_t* data, size_t len,
                                                         void* ctx);
 
// phase-402: the trailing `sched_context` / `callback_group` arguments moved
// into `nros_cpp_subscription_options_t` (defined by the cbindgen output above);
// a NULL `options` is all-defaults, i.e. the pre-phase-402 behaviour.
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);
 
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);
 
// Phase 189.M3.4 — callback-style register that also delivers the sample's wire
// attachment (5-arg trampoline). Same cbindgen-exclusion reason as above.
typedef 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);
 
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);
// Phase 269 W3 — callback-style subscription that delivers the sample's E2E
// integrity status alongside the CDR bytes. Same cbindgen-exclusion reason as
// above (takes `RawSubscriptionSafetyCallback`, an external nros-node alias
// gated on the `safety-e2e` feature). The 6-arg trampoline unpacks the three
// integrity scalars; `subscription.hpp` repacks them into
// `nros_cpp_integrity_status_t` for the typed user handler. Gated on the
// `NROS_SAFETY_E2E` build feature (lowered from `[system].features =
// ["safety"]` via `NanoRosCapabilities.cmake`).
#if defined(NROS_SAFETY_E2E)
typedef void (*nros_cpp_subscription_validated_callback_t)(const uint8_t* data, size_t len,
                                                           int64_t gap, bool duplicate,
                                                           int8_t crc_valid, void* ctx);
 
nros_cpp_ret_t nros_cpp_subscription_register_validated(
    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_validated_callback_t callback, void* context,
    size_t* out_handle_id, const nros_cpp_subscription_options_t* options);
#endif // NROS_SAFETY_E2E
} // extern "C"
 
namespace rclcpp {
 
using IntegrityStatus = ::nros_cpp_integrity_status_t;
 
static constexpr size_t SUBSCRIPTION_TOPIC_NAME_MAX = 256;
 
} // namespace rclcpp
 
namespace rclcpp {
class Node;
}
 
// ============================================================================
// `rclcpp::Subscription<M>` -- DEFINED here (RFC-0089: rclcpp:: is the home)
// ============================================================================
//
// phase-428: the definition moved from `nros::` to `rclcpp::` (the `nros::`
// alias went in phase-483 W1). The nested `SharedPtr` / `ConstSharedPtr` / `UniquePtr`
// aliases live on the class itself, so `rclcpp::Subscription<M>::SharedPtr`
// resolves with no wrapper type in between.
namespace rclcpp {
 
template <typename M> class Subscription {
  public:
    using SharedPtr = ::rclcpp::SubscriptionHandle<M>;
    using ConstSharedPtr = ::rclcpp::SubscriptionHandle<M>;
    using UniquePtr = ::rclcpp::SubscriptionHandle<M>;
 
    using TypedSubscriptionFn = void (*)(const M& msg);
    using TypedSubscriptionFnWithCtx = void (*)(const M& msg, void* ctx);
    // Phase 189.M3.4 — callback-with-attachment handler (`bridge_origin` etc.).
    using TypedSubscriptionInfoFn = void (*)(const M& msg, const uint8_t* attachment,
                                             size_t attachment_len);
    // Phase 269 W3 — callback-with-integrity handler: receives the deserialized
    // message plus the sample's E2E CRC/sequence status. Requires
    // `NROS_SAFETY_E2E` (lowered from `[system].features = ["safety"]`).
#if defined(NROS_SAFETY_E2E)
    using TypedSubscriptionSafetyFn = void (*)(const M& msg,
                                               const ::rclcpp::IntegrityStatus& integrity);
#endif // NROS_SAFETY_E2E
 
    const char* get_topic_name() const { return initialized_ ? topic_name_ : ""; }
 
    bool is_valid() const { return initialized_; }
 
    Result get_publisher_count(::rclcpp::Executor& executor, size_t* out_count) const {
        if (!initialized_) return Result(::rclcpp::ErrorCode::NotInitialized);
        return executor.count_publishers(topic_name_, out_count);
    }
 
    ~Subscription() { initialized_ = false; }
 
    // Move semantics (non-copyable). Bookkeeping only — there is no storage to
    // relocate, so `nros_cpp_subscription_relocate` is not called here.
    //
    // THE HAZARD, narrowed to its real subject: the out-ref creators below hand
    // the arena `&out` as the trampoline context, so an object registered that
    // way must NOT be moved afterwards — the move copies the handler pointers
    // and leaves the arena dispatching into the source object's address. The
    // returning `create_subscription` (`nros.hpp`) has no such hazard: W1 put
    // the capture in the arena and it hands back a `SharedPtr`, which is a
    // value with nothing pointing at it.
    Subscription(Subscription&& other) : initialized_(other.initialized_) {
        ::memcpy(topic_name_, other.topic_name_, sizeof(topic_name_));
        user_fn_ = other.user_fn_;
        user_fn_ctx_ = other.user_fn_ctx_;
        user_ctx_ = other.user_ctx_;
        user_fn_info_ = other.user_fn_info_;
#if defined(NROS_SAFETY_E2E)
        user_fn_safety_ = other.user_fn_safety_;
#endif // NROS_SAFETY_E2E
        sched_handle_id_ = other.sched_handle_id_;
        executor_ = other.executor_;
        other.initialized_ = false;
    }
 
    Subscription& operator=(Subscription&& other) {
        if (this != &other) {
            initialized_ = other.initialized_;
            ::memcpy(topic_name_, other.topic_name_, sizeof(topic_name_));
            user_fn_ = other.user_fn_;
            user_fn_ctx_ = other.user_fn_ctx_;
            user_ctx_ = other.user_ctx_;
            user_fn_info_ = other.user_fn_info_;
#if defined(NROS_SAFETY_E2E)
            user_fn_safety_ = other.user_fn_safety_;
#endif // NROS_SAFETY_E2E
            sched_handle_id_ = other.sched_handle_id_;
            executor_ = other.executor_;
            other.initialized_ = false;
        }
        return *this;
    }
 
    Subscription() : topic_name_{}, initialized_(false) {}
 
    bool has_sched_handle() const { return sched_handle_id_ != static_cast<size_t>(-1); }
    size_t sched_handle_id() const { return sched_handle_id_; }
 
    ::rclcpp::QoS get_actual_qos() const {
        nros_cpp_qos_t f{};
        if (!initialized_ || executor_ == nullptr ||
            nros_cpp_subscription_get_actual_qos(nullptr, executor_, sched_handle_id_, &f) != 0) {
            return ::rclcpp::detail::qos_all_unknown();
        }
        return ::rclcpp::detail::qos_from_ffi(f);
    }
 
  private:
    Subscription(const Subscription&) = delete;
    Subscription& operator=(const Subscription&) = delete;
 
    friend class ::rclcpp::Node;
 
    static void message_trampoline(const uint8_t* data, size_t len, void* ctx) {
        auto* self = static_cast<Subscription*>(ctx);
        if (self == nullptr) return;
        M msg;
        if (M::ffi_deserialize(data, len, &msg) != 0) return;
        if (self->user_fn_ != nullptr) {
            self->user_fn_(msg);
        } else if (self->user_fn_ctx_ != nullptr) {
            self->user_fn_ctx_(msg, self->user_ctx_);
        }
    }
 
    static void message_info_trampoline(const uint8_t* data, size_t len, const uint8_t* attachment,
                                        size_t attachment_len, void* ctx) {
        auto* self = static_cast<Subscription*>(ctx);
        if (self == nullptr) return;
        M msg;
        if (M::ffi_deserialize(data, len, &msg) != 0) return;
        if (self->user_fn_info_ != nullptr) {
            self->user_fn_info_(msg, attachment, attachment_len);
        }
    }
 
#if defined(NROS_SAFETY_E2E)
    static void message_safety_trampoline(const uint8_t* data, size_t len, int64_t gap,
                                          bool duplicate, int8_t crc_valid, void* ctx) {
        auto* self = static_cast<Subscription*>(ctx);
        if (self == nullptr) return;
        M msg;
        if (M::ffi_deserialize(data, len, &msg) != 0) return;
        if (self->user_fn_safety_ != nullptr) {
            nros_cpp_integrity_status_t status;
            status.gap = gap;
            status.duplicate = duplicate;
            status.crc_valid = crc_valid;
            self->user_fn_safety_(msg, status);
        }
    }
#endif // NROS_SAFETY_E2E
 
    void store_topic_name(const char* topic) {
        size_t n = 0;
        while (topic[n] != '\0' && n + 1 < sizeof(topic_name_)) {
            topic_name_[n] = topic[n];
            ++n;
        }
        topic_name_[n] = '\0';
    }
 
    // phase-456 W2b — `storage_` (NROS_SUBSCRIBER_SIZE bytes) and `stream_` are
    // GONE, with the taking API that was their only reader. On this path the
    // arena owns the subscriber, so those bytes were never filled, and
    // `take()` reinterpreted them (`&mut *(storage as *mut RmwSubscriber)`)
    // rather than refusing. `callback_mode_` went with them: it distinguished
    // the two owners inside one class, and there is one owner here now.
    char topic_name_[::rclcpp::SUBSCRIPTION_TOPIC_NAME_MAX];
    bool initialized_;
    // Phase 189.M3.1 — executor HandleId for sched-context binding, or
    // SIZE_MAX (the default) when the registration has not happened yet.
    size_t sched_handle_id_ = static_cast<size_t>(-1);
    // The executor that owns the arena entry `sched_handle_id_` indexes —
    // the phase-467 RMW gap-closure design study's Row 13.
    //
    // A `HandleId` is EXECUTOR-SCOPED: an index into one executor's arena, not
    // a process-wide identity, and RFC-0002 puts one executor on one RTOS
    // task, so a tiered image has several. That is why `get_actual_qos()`
    // could not be answered from `sched_handle_id_` alone, and why resolving
    // it "from the sched handle" would have needed a global executor
    // registry. The pair is the same one `rclcpp::Client` already keeps
    // (`{executor_, handle_id_}`), for the same reason.
    void* executor_ = nullptr;
    // The user's handler, dispatched by `message_trampoline` during spin.
    TypedSubscriptionFn user_fn_ = nullptr;
    TypedSubscriptionFnWithCtx user_fn_ctx_ = nullptr;
    TypedSubscriptionInfoFn user_fn_info_ = nullptr;
    void* user_ctx_ = nullptr;
#if defined(NROS_SAFETY_E2E)
    // Phase 269 W3 — handler for the integrity-carrying callback path; nullptr
    // when not using `create_subscription_with_safety`.
    TypedSubscriptionSafetyFn user_fn_safety_ = nullptr;
#endif // NROS_SAFETY_E2E
};
 
} // namespace rclcpp
 
// Phase 84.G8: out-of-line definition of Node::create_subscription<M>().
#include "rclcpp/node.hpp"
 
namespace rclcpp {
 
// Phase 189.M3.x — callback-style (arena-registered) subscription. The arena
// owns the subscriber + dispatches `out`'s message handler during spin_once, so
// the handle is real and `options.sched_context` is functional. Mirrors the
// callback-style `create_service` one entity over.
} // namespace rclcpp
 
namespace rclcpp {
template <typename M, typename F, typename>
Result Node::create_subscription(Subscription<M>& out, const char* topic, const ::rclcpp::QoS& qos,
                                 F callback, const ::rclcpp::SubscriptionOptions& options) {
    // RFC-0088 D5 — one image, one backend, one encoding. Compile-time, so a
    // message the linked backend cannot encode never reaches the wire.
    NROS_CPP_ASSERT_MESSAGE_FORMAT(M);
    if (!initialized_) return Result(::rclcpp::ErrorCode::NotInitialized);
    nros_cpp_qos_t ffi_qos = ::rclcpp::detail::qos_to_ffi(qos);
 
    // Store the user handler (compile error if F isn't convertible to the
    // plain-fn-ptr handler type).
    out.user_fn_ = typename Subscription<M>::TypedSubscriptionFn(callback);
    out.user_fn_ctx_ = nullptr;
    out.user_ctx_ = nullptr;
 
    uint8_t sched = (options.sched_context == ::rclcpp::SCHED_CONTEXT_UNSET)
                        ? 0u
                        : static_cast<uint8_t>(options.sched_context);
    size_t handle = static_cast<size_t>(-1);
    // phase-402: `sched_context` is a FIELD now. `callback_group` stays unset,
    // i.e. the default group.
    nros_cpp_subscription_options_t ffi_options = nros_cpp_subscription_default_options();
    ffi_options.sched_context = sched;
    // phase-456 W7 — state the bound. `M` is a template parameter of this
    // function, so the number was always available here; leaving it at the
    // option default put this registration on issue 1319's `c_raw_no_hint` row
    // while the sizing descriptor credited the entry with a supplied hint.
    ffi_options.rx_buffer_hint = static_cast<uint32_t>(::rclcpp::rx_buffer_capacity<M>::value);
    nros_cpp_ret_t ret = nros_cpp_subscription_register(
        &handle_, topic, M::TYPE_NAME, M::TYPE_HASH, ffi_qos, &Subscription<M>::message_trampoline,
        &out, &handle, &ffi_options);
    if (ret == 0) {
        out.sched_handle_id_ = handle;
        out.executor_ = executor_handle_;
        // phase-456 W2b — `get_topic_name()` answered "" on every callback-style
        // subscription until this line existed.
        out.store_topic_name(topic);
        out.initialized_ = true;
    }
    return Result(ret);
}
} // namespace rclcpp
 
namespace rclcpp {
 
// Phase 273 (RFC-0047) — callback-style subscription **in** a named callback group.
// Mirrors create_subscription (callback-style) exactly but passes group.get_name()
// as `callback_group` so the executor binds the slot via group_sched_table.
} // namespace rclcpp
 
namespace rclcpp {
template <typename M, typename F, typename>
Result Node::create_subscription_in_group(const ::rclcpp::CallbackGroup& group,
                                          Subscription<M>& out, const char* topic,
                                          const ::rclcpp::QoS& qos, F callback,
                                          const ::rclcpp::SubscriptionOptions& options) {
    // RFC-0088 D5 — one image, one backend, one encoding. Compile-time, so a
    // message the linked backend cannot encode never reaches the wire.
    NROS_CPP_ASSERT_MESSAGE_FORMAT(M);
    if (!initialized_) return Result(::rclcpp::ErrorCode::NotInitialized);
    nros_cpp_qos_t ffi_qos = ::rclcpp::detail::qos_to_ffi(qos);
 
    out.user_fn_ = typename Subscription<M>::TypedSubscriptionFn(callback);
    out.user_fn_ctx_ = nullptr;
    out.user_ctx_ = nullptr;
 
    uint8_t sched = (options.sched_context == ::rclcpp::SCHED_CONTEXT_UNSET)
                        ? 0u
                        : static_cast<uint8_t>(options.sched_context);
    size_t handle = static_cast<size_t>(-1);
    // phase-402: both the sched slot and the Phase 273 group name are FIELDS now.
    nros_cpp_subscription_options_t ffi_options = nros_cpp_subscription_default_options();
    ffi_options.sched_context = sched;
    ffi_options.callback_group = group.get_name();
    // phase-456 W7 — see the ungrouped form above. A grouped subscription costs
    // the arena exactly what an ungrouped one does, so it states the same bound.
    ffi_options.rx_buffer_hint = static_cast<uint32_t>(::rclcpp::rx_buffer_capacity<M>::value);
    nros_cpp_ret_t ret = nros_cpp_subscription_register(
        &handle_, topic, M::TYPE_NAME, M::TYPE_HASH, ffi_qos, &Subscription<M>::message_trampoline,
        &out, &handle, &ffi_options);
    if (ret == 0) {
        out.sched_handle_id_ = handle;
        out.executor_ = executor_handle_;
        // phase-456 W2b — `get_topic_name()` answered "" on every callback-style
        // subscription until this line existed.
        out.store_topic_name(topic);
        out.initialized_ = true;
    }
    return Result(ret);
}
} // namespace rclcpp
 
namespace rclcpp {
 
// Phase 189.M3.4 — callback-style subscription that delivers the wire attachment.
// Mirrors the callback `create_subscription` one step over, but stores the
// `(const M&, attachment, att_len)` handler + registers via the with-info arena
// path so the trampoline receives the attachment.
} // namespace rclcpp
 
namespace rclcpp {
template <typename M, typename F, typename>
Result Node::create_subscription_with_info(Subscription<M>& out, const char* topic,
                                           const ::rclcpp::QoS& qos, F callback,
                                           const ::rclcpp::SubscriptionOptions& options) {
    // RFC-0088 D5 — one image, one backend, one encoding. Compile-time, so a
    // message the linked backend cannot encode never reaches the wire.
    NROS_CPP_ASSERT_MESSAGE_FORMAT(M);
    if (!initialized_) return Result(::rclcpp::ErrorCode::NotInitialized);
    nros_cpp_qos_t ffi_qos = ::rclcpp::detail::qos_to_ffi(qos);
 
    out.user_fn_info_ = typename Subscription<M>::TypedSubscriptionInfoFn(callback);
    out.user_fn_ = nullptr;
    out.user_fn_ctx_ = nullptr;
    out.user_ctx_ = nullptr;
 
    uint8_t sched = (options.sched_context == ::rclcpp::SCHED_CONTEXT_UNSET)
                        ? 0u
                        : static_cast<uint8_t>(options.sched_context);
    size_t handle = static_cast<size_t>(-1);
    // phase-402: `sched_context` is a FIELD now.
    nros_cpp_subscription_options_t ffi_options = nros_cpp_subscription_default_options();
    ffi_options.sched_context = sched;
    // phase-456 W7 — the attachment rides beside the sample and does not change
    // how many bytes the sample itself needs, so this is the same bound the
    // plain callback form states.
    ffi_options.rx_buffer_hint = static_cast<uint32_t>(::rclcpp::rx_buffer_capacity<M>::value);
    nros_cpp_ret_t ret = nros_cpp_subscription_register_with_info(
        &handle_, topic, M::TYPE_NAME, M::TYPE_HASH, ffi_qos,
        &Subscription<M>::message_info_trampoline, &out, &handle, &ffi_options);
    if (ret == 0) {
        out.sched_handle_id_ = handle;
        out.executor_ = executor_handle_;
        // phase-456 W2b — `get_topic_name()` answered "" on every callback-style
        // subscription until this line existed.
        out.store_topic_name(topic);
        out.initialized_ = true;
    }
    return Result(ret);
}
} // namespace rclcpp
 
namespace rclcpp {
 
#if defined(NROS_SAFETY_E2E)
// Phase 269 W3 — out-of-line definition of Node::create_subscription_with_safety.
// Mirrors `create_subscription_with_info` one overload over, but routes through
// `nros_cpp_subscription_register_validated` so the arena dispatches the
// `message_safety_trampoline` on each new sample.
} // namespace rclcpp
 
namespace rclcpp {
template <typename M, typename F, typename>
Result Node::create_subscription_with_safety(Subscription<M>& out, const char* topic,
                                             const ::rclcpp::QoS& qos, F callback,
                                             const ::rclcpp::SubscriptionOptions& options) {
    // RFC-0088 D5 — one image, one backend, one encoding. Compile-time, so a
    // message the linked backend cannot encode never reaches the wire.
    NROS_CPP_ASSERT_MESSAGE_FORMAT(M);
    if (!initialized_) return Result(::rclcpp::ErrorCode::NotInitialized);
    nros_cpp_qos_t ffi_qos = ::rclcpp::detail::qos_to_ffi(qos);
 
    out.user_fn_safety_ = typename Subscription<M>::TypedSubscriptionSafetyFn(callback);
    out.user_fn_ = nullptr;
    out.user_fn_ctx_ = nullptr;
    out.user_fn_info_ = nullptr;
    out.user_ctx_ = nullptr;
 
    uint8_t sched = (options.sched_context == ::rclcpp::SCHED_CONTEXT_UNSET)
                        ? 0u
                        : static_cast<uint8_t>(options.sched_context);
    size_t handle = static_cast<size_t>(-1);
    // phase-402: `sched_context` is a FIELD now.
    nros_cpp_subscription_options_t ffi_options = nros_cpp_subscription_default_options();
    ffi_options.sched_context = sched;
    // phase-456 W7 — the validated path deserializes the same bytes; the safety
    // status is computed from them, not received alongside them.
    ffi_options.rx_buffer_hint = static_cast<uint32_t>(::rclcpp::rx_buffer_capacity<M>::value);
    nros_cpp_ret_t ret = nros_cpp_subscription_register_validated(
        &handle_, topic, M::TYPE_NAME, M::TYPE_HASH, ffi_qos,
        &Subscription<M>::message_safety_trampoline, &out, &handle, &ffi_options);
    if (ret == 0) {
        out.sched_handle_id_ = handle;
        out.executor_ = executor_handle_;
        // phase-456 W2b — `get_topic_name()` answered "" on every callback-style
        // subscription until this line existed.
        out.store_topic_name(topic);
        out.initialized_ = true;
    }
    return Result(ret);
}
} // namespace rclcpp
 
namespace rclcpp {
#endif // NROS_SAFETY_E2E
 
namespace detail {
 
template <typename M, typename Fn>
inline Result register_subscription_capturing(::rclcpp::Node& node, const char* topic,
                                              const QoS& qos, const Fn& fn, size_t* out_handle_id) {
    // phase-456 W8 — the per-entry capture budget this used to assert is GONE.
    // It read `sizeof(Fn) <= NROS_CPP_CALLBACK_CAPACITY + 2 * sizeof(void*)` and
    // named `CALLBACK_CAPTURE_BYTES`, an arena constant that no longer exists:
    // the runtime allocates exactly `capture_len` bytes now, so there is no
    // cross-language number for the two sides to agree about and nothing here
    // to assert. `InplaceFn`'s own capacity `static_assert` is untouched and is
    // still the knob a too-large CALLABLE hits (`inplace_fn.hpp`), which is a
    // property of that type rather than of this registration.
    const nros_cpp_node_t* h = node.ffi_handle();
    if (h == nullptr) return Result(ErrorCode::NotInitialized);
 
    nros_cpp_subscription_message_callback_t invoke = [](const uint8_t* data, size_t len,
                                                         void* ctx) {
        // `ctx` is the ARENA's copy of `fn`, not ours.
        const Fn* self = static_cast<const Fn*>(ctx);
        M msg;
        if (M::ffi_deserialize(data, len, &msg) != 0) return;
        (*self)(msg);
    };
 
    // phase-476 W2 — the arena's copy is destroyed when the subscription is
    // released. `Fn` is an `rclcpp::InplaceFn` (the one caller passes one), so
    // `clear()` runs the stored callable's own destructor.
    void (*destroy)(void*) = [](void* ctx) { static_cast<Fn*>(ctx)->clear(); };
 
    nros_cpp_qos_t ffi_qos = detail::qos_to_ffi(qos);
    nros_cpp_subscription_options_t opts = {};
    opts.rx_buffer_hint = static_cast<uint32_t>(::rclcpp::rx_buffer_capacity<M>::value);
    return Result(nros_cpp_subscription_register_capturing(
        h, topic, M::TYPE_NAME, M::TYPE_HASH, ffi_qos, invoke,
        reinterpret_cast<const uint8_t*>(&fn), sizeof(Fn), destroy, out_handle_id, &opts));
}
 
} // namespace detail
 
} // namespace rclcpp
 
// ============================================================================
// WHAT LEFT THIS HEADER
// ============================================================================
//
// phase-456 W2 deleted `rclcpp::detail::SubscriptionCallback<M>` — the heap
// cell whose only job was to give a capturing lambda a stable address. W1 put
// the capture in the arena entry, so the cell had nothing left to hold, and
// `nros.hpp`'s returning `create_subscription` stopped allocating one.
//
// phase-456 W2b moved the TAKING API — `take`, `take_sized`, `take_serialized`,
// `take_serialized_with_attachment`, `take_validated`, `take_validated_sized`,
// `take_sequence`, `try_borrow` + `View`, `stream()` and the three QoS-event setters — to
// `rclcpp::PollSubscription<M>` in `rclcpp/polling_subscription.hpp`, together with
// the out-ref creator that fills it. Those calls all dereference `storage_`,
// which only a caller-owned subscriber ever has.
 
#endif // NROS_RCLCPP_SUBSCRIPTION_HPP