publisher.hpp

Go to the documentation of this file.

// nros-cpp: Publisher class
// Freestanding C++ — no exceptions, no STL required
 
#ifndef NROS_CPP_PUBLISHER_HPP
#define NROS_CPP_PUBLISHER_HPP
 
#include <cstdint>
#include <cstddef>
#include <string.h> // memcpy — `<cstring>` isn't in Zephyr's minimal libcpp
 
#include "nros/config.hpp"
#include "nros/result.hpp"
 
#include "nros_cpp_ffi.h"
 
namespace nros {
 
static constexpr size_t PUBLISHER_TOPIC_NAME_MAX = 256;
 
template <typename M> class Publisher {
  public:
    Result publish(const M& msg) { return Result(M::ffi_publish(storage_, &msg)); }
 
    Result publish_raw(const uint8_t* data, size_t len) {
        if (!initialized_) return Result(ErrorCode::NotInitialized);
        return Result(nros_cpp_publish_raw(storage_, data, len));
    }
 
    template <typename W> Result publish_streamed(size_t total_len, W&& writer) {
        if (!initialized_) return Result(ErrorCode::NotInitialized);
        struct Ctx {
            W writer;
            size_t total;
        } ctx{static_cast<W&&>(writer), total_len};
        auto size_cb = [](size_t* out_total, void* user) {
            auto* c = static_cast<Ctx*>(user);
            *out_total = c->total;
        };
        auto chunk_cb = [](uint8_t* out_buf, size_t cap, size_t* out_written, void* user) {
            auto* c = static_cast<Ctx*>(user);
            *out_written = c->writer(out_buf, cap);
        };
        return Result(nros_cpp_publisher_publish_streamed(storage_, size_cb, chunk_cb,
                                                          static_cast<void*>(&ctx)));
    }
 
    // ====================================================================
    // Phase 124.A.7 — zero-copy publish (loan / commit / discard)
    // ====================================================================
 
    class Loan {
      public:
        Loan() : pub_(nullptr), buf_(nullptr), cap_(0), token_(nullptr) {}
        Loan(Loan&& o) : pub_(o.pub_), buf_(o.buf_), cap_(o.cap_), token_(o.token_) {
            o.pub_ = nullptr;
            o.token_ = nullptr;
        }
        Loan& operator=(Loan&& o) {
            if (this != &o) {
                release();
                pub_ = o.pub_;
                buf_ = o.buf_;
                cap_ = o.cap_;
                token_ = o.token_;
                o.pub_ = nullptr;
                o.token_ = nullptr;
            }
            return *this;
        }
        Loan(const Loan&) = delete;
        Loan& operator=(const Loan&) = delete;
        ~Loan() { release(); }
 
        uint8_t* data() { return buf_; }
        const uint8_t* data() const { return buf_; }
        size_t capacity() const { return cap_; }
 
        bool is_valid() const { return token_ != nullptr; }
 
        Result commit(size_t actual_len) {
            if (!token_) return Result(ErrorCode::NotInitialized);
            nros_cpp_ret_t ret = nros_cpp_publisher_commit(pub_, token_, actual_len);
            pub_ = nullptr;
            token_ = nullptr;
            buf_ = nullptr;
            cap_ = 0;
            return Result(ret);
        }
 
        Result discard() {
            if (!token_) return Result::success();
            nros_cpp_ret_t ret = nros_cpp_publisher_discard(pub_, token_);
            pub_ = nullptr;
            token_ = nullptr;
            buf_ = nullptr;
            cap_ = 0;
            return Result(ret);
        }
 
        Loan(void* pub, uint8_t* buf, size_t cap, void* token)
            : pub_(pub), buf_(buf), cap_(cap), token_(token) {}
 
      private:
        void release() {
            if (token_ && pub_) {
                nros_cpp_publisher_discard(pub_, token_);
                token_ = nullptr;
            }
        }
 
        void* pub_;
        uint8_t* buf_;
        size_t cap_;
        void* token_;
    };
 
    Expected<Loan> loan(size_t requested_len) {
        if (!initialized_) return Expected<Loan>::error(Result(ErrorCode::NotInitialized));
        uint8_t* buf = nullptr;
        size_t cap = 0;
        void* token = nullptr;
        nros_cpp_ret_t ret = nros_cpp_publisher_loan(storage_, requested_len, &buf, &cap, &token);
        if (ret != 0) return Expected<Loan>::error(Result(ret));
        return Expected<Loan>::ok(Loan{storage_, buf, cap, token});
    }
 
    const char* get_topic_name() const { return initialized_ ? topic_name_ : ""; }
 
    bool is_valid() const { return initialized_; }
 
    ~Publisher() {
        if (initialized_) {
            nros_cpp_publisher_destroy(storage_);
            initialized_ = false;
        }
    }
 
    // Move semantics (non-copyable). Relocation goes through the
    // `nros_cpp_publisher_relocate` runtime call.
    Publisher(Publisher&& other) : initialized_(other.initialized_) {
        if (other.initialized_) {
            nros_cpp_publisher_relocate(other.storage_, storage_);
            ::memcpy(topic_name_, other.topic_name_, sizeof(topic_name_));
            other.initialized_ = false;
        }
    }
 
    Publisher& operator=(Publisher&& other) {
        if (this != &other) {
            if (initialized_) {
                nros_cpp_publisher_destroy(storage_);
                initialized_ = false;
            }
            if (other.initialized_) {
                nros_cpp_publisher_relocate(other.storage_, storage_);
                ::memcpy(topic_name_, other.topic_name_, sizeof(topic_name_));
                initialized_ = true;
                other.initialized_ = false;
            }
        }
        return *this;
    }
 
    Publisher() : storage_(), topic_name_{}, initialized_(false) {}
 
    // ====================================================================
    // Phase 108 — status events
    // ====================================================================
 
    Result on_liveliness_lost(nros_cpp_publisher_count_cb_t cb, void* user_context = nullptr) {
        if (!initialized_) return Result(ErrorCode::NotInitialized);
        return Result(nros_cpp_publisher_set_liveliness_lost(storage_, cb, user_context));
    }
 
    Result on_offered_deadline_missed(uint32_t deadline_ms, nros_cpp_publisher_count_cb_t cb,
                                      void* user_context = nullptr) {
        if (!initialized_) return Result(ErrorCode::NotInitialized);
        return Result(nros_cpp_publisher_set_offered_deadline_missed(storage_, deadline_ms, cb,
                                                                     user_context));
    }
 
    Result assert_liveliness() {
        if (!initialized_) return Result(ErrorCode::NotInitialized);
        return Result(nros_cpp_publisher_assert_liveliness(storage_));
    }
 
  private:
    Publisher(const Publisher&) = delete;
    Publisher& operator=(const Publisher&) = delete;
 
    friend class Node;
 
    alignas(8) uint8_t storage_[NROS_PUBLISHER_SIZE];
    char topic_name_[PUBLISHER_TOPIC_NAME_MAX];
    bool initialized_;
};
 
} // namespace nros
 
// Phase 84.G8: out-of-line definition of Node::create_publisher<M>().
// Placed here so consumers only pay for this template when they actually
// use publishers — including `nros/node.hpp` alone no longer drags in
// every entity's code path.
#include "nros/node.hpp"
 
namespace nros {
 
template <typename M>
Result Node::create_publisher(Publisher<M>& out, const char* topic, const QoS& qos) {
    if (!initialized_) return Result(ErrorCode::NotInitialized);
    nros_cpp_qos_t ffi_qos;
    ffi_qos.reliability = static_cast<nros_cpp_qos_reliability_t>(qos.reliability_raw());
    ffi_qos.durability = static_cast<nros_cpp_qos_durability_t>(qos.durability_raw());
    ffi_qos.history = static_cast<nros_cpp_qos_history_t>(qos.history_raw());
    ffi_qos.liveliness_kind = static_cast<nros_cpp_qos_liveliness_t>(qos.liveliness_raw());
    ffi_qos.depth = qos.depth();
    ffi_qos.deadline_ms = qos.deadline_ms();
    ffi_qos.lifespan_ms = qos.lifespan_ms();
    ffi_qos.liveliness_lease_ms = qos.liveliness_lease_ms();
    ffi_qos.avoid_ros_namespace_conventions = qos.avoid_ros_namespace_conventions() ? 1 : 0;
    nros_cpp_ret_t ret = nros_cpp_publisher_create(&handle_, topic, M::TYPE_NAME, M::TYPE_HASH,
                                                   ffi_qos, out.storage_);
    if (ret == 0) {
        // Topic name lives C++-side; copy + null-terminate into the
        // fixed-size buffer. Truncation is silent.
        size_t topic_len = 0;
        while (topic[topic_len] != '\0' && topic_len + 1 < sizeof(out.topic_name_)) {
            out.topic_name_[topic_len] = topic[topic_len];
            ++topic_len;
        }
        out.topic_name_[topic_len] = '\0';
        out.initialized_ = true;
    }
    return Result(ret);
}
 
template <typename M>
Result Node::create_publisher(Publisher<M>& out, const char* topic, const QoS& qos,
                              const PublisherOptions& options) {
    (void)options; // reserved — no live fields today
    return create_publisher<M>(out, topic, qos);
}
 
template <typename M>
inline Expected<Publisher<M>> make_publisher(Node& node, const char* topic,
                                             const QoS& qos = QoS::default_profile()) {
    Publisher<M> p;
    Result r = node.create_publisher<M>(p, topic, qos);
    if (!r.ok()) return Expected<Publisher<M>>::error(r);
    return Expected<Publisher<M>>::ok(std::move(p));
}
 
} // namespace nros
 
#endif // NROS_CPP_PUBLISHER_HPP