Aller au contenu principal

C++ API Reference

HDDS provides a modern C++ SDK with RAII wrappers around the C FFI layer. The API uses C++17 features for safe, idiomatic usage.

Version 1.0.0

This documents the current v1.0.0 API. Listeners are supported via hdds_listener.hpp. Instance management and content-filtered topics are not yet implemented.

Installation​

# Build from source
cd /path/to/hdds/sdk/cxx
mkdir build && cd build
cmake ..
make -j$(nproc)

# CMake integration
find_package(hdds REQUIRED)
target_link_libraries(myapp hdds::hdds)
#include <hdds.hpp>

Quick Start​

#include <hdds.hpp>

int main() {
    // RAII participant
    hdds::Participant participant("my_app");

    // Fluent QoS builder
    auto qos = hdds::QoS::reliable()
        .transient_local()
        .history_depth(10);

    // Create writer (raw bytes)
    auto writer = participant.create_writer_raw("hello/world", qos);

    // Publish
    std::vector<uint8_t> data = {1, 2, 3, 4};
    writer->write_raw(data);

    return 0;  // RAII cleanup
}

Participant​

Entry point for all DDS operations.

Creation​

// Basic creation (domain 0)
hdds::Participant participant("my_app");

// With domain ID
hdds::Participant participant("my_app", 42);

Properties​

const std::string& name = participant.name();
uint32_t domain = participant.domain_id();
uint8_t pid = participant.participant_id();

// Get C handle for advanced usage
HddsParticipant* c_handle = participant.c_handle();

Creating Writers/Readers​

// Raw (untyped) writer/reader
auto writer = participant.create_writer_raw("topic", qos);
auto reader = participant.create_reader_raw("topic", qos);

// Typed writer/reader (requires CDR2 serialization)
auto writer = participant.create_writer<MyType>("topic", qos);
auto reader = participant.create_reader<MyType>("topic", qos);

// Publisher/Subscriber grouping
auto publisher = participant.create_publisher(qos);
auto subscriber = participant.create_subscriber(qos);

DSCP Configuration​

Not Yet Available in C++ SDK

DSCP configuration is implemented in the Rust core but the C++ SDK bindings (set_dscp(), DscpConfig::from_env(), DscpConfig::class_name()) are not yet exported. Use the HDDS_DSCP environment variable or configure via the Rust API directly.

DSCP classes (available in Rust and C APIs):

enum class DscpClass : uint8_t {
    BestEffort = 0,   // CS0 - Default, no priority
    Af11 = 10,        // High-throughput data
    Af21 = 18,        // Low-latency data (standard DDS)
    Af31 = 26,        // Streaming media
    Af41 = 34,        // Video, important telemetry
    Ef = 46,          // Real-time, safety-critical
    Cs6 = 48,         // Network control
    Cs7 = 56,         // Highest priority
};

QoS Configuration​

Fluent builder API for Quality of Service.

Factory Methods​

// Predefined profiles
auto qos = hdds::QoS::default_qos();   // BestEffort, Volatile
auto qos = hdds::QoS::reliable();       // Reliable delivery
auto qos = hdds::QoS::best_effort();    // Fire and forget
auto qos = hdds::QoS::rti_defaults();   // RTI Connext compatible

// Load from XML file
auto qos = hdds::QoS::from_file("fastdds_profile.xml");

Fluent Builder​

auto qos = hdds::QoS::reliable()
    .transient_local()                          // Durability
    .history_depth(100)                         // History
    .deadline(std::chrono::milliseconds(100))   // Deadline
    .lifespan(std::chrono::seconds(5))          // Lifespan
    .liveliness_automatic(std::chrono::seconds(1))
    .ownership_exclusive(100)                   // Ownership strength
    .partition("sensors")                       // Partition
    .time_based_filter(std::chrono::milliseconds(10))
    .transport_priority(10)
    .resource_limits(1000, 100, 10);            // max_samples, instances, per_instance

Inspection​

bool reliable = qos.is_reliable();
bool transient = qos.is_transient_local();
uint32_t depth = qos.get_history_depth();

// Get C handle for FFI
HddsQoS* c_handle = qos.c_handle();

DataWriter​

Writers publish data to a topic.

Creation​

// Create with default QoS
auto writer = participant.create_writer_raw("topic");

// Create with custom QoS
auto qos = hdds::QoS::reliable().transient_local();
auto writer = participant.create_writer_raw("topic", qos);

Writing Data​

// Write raw bytes (vector)
std::vector<uint8_t> data = {1, 2, 3, 4};
writer->write_raw(data);

// Write raw bytes (pointer + size)
uint8_t buffer[256];
writer->write_raw(buffer, sizeof(buffer));

// Typed write (requires T with CDR2 serialization)
MyType msg{.value = 42};
writer->write(msg);

Properties​

const std::string& topic = writer->topic_name();

Move Semantics​

// Writers are movable, not copyable
auto writer2 = std::move(writer);  // OK
// auto writer3 = writer2;          // Error: deleted copy constructor

DataReader​

Readers receive data from a topic.

Creation​

// Create with default QoS
auto reader = participant.create_reader_raw("topic");

// Create with custom QoS
auto qos = hdds::QoS::reliable().history_depth(100);
auto reader = participant.create_reader_raw("topic", qos);

Taking Data​

// Take raw bytes (non-blocking)
std::optional<std::vector<uint8_t>> data = reader->take_raw();
if (data) {
    // Process data
    process(*data);
}

// Typed take
std::optional<MyType> msg = reader->take<MyType>();
if (msg) {
    std::cout << "Received: " << msg->value << "\n";
}

Status Condition​

// Get status condition for WaitSet
HddsStatusCondition* cond = reader->get_status_condition();

WaitSet​

Event-driven waiting for data availability.

Basic Usage​

// Create waitset
hdds::WaitSet waitset;

// Create reader and attach condition
auto reader = participant.create_reader_raw("topic");
auto* cond = reader->get_status_condition();
waitset.attach(cond);

// Wait loop
while (running) {
    bool triggered = waitset.wait(std::chrono::seconds(1));
    if (triggered) {
        while (auto data = reader->take_raw()) {
            process(*data);
        }
    }
}

// Cleanup
waitset.detach(cond);

Guard Conditions​

// Create guard condition for custom signaling
hdds::GuardCondition guard;
waitset.attach(guard);

// Trigger from another thread
std::thread([&guard]() {
    std::this_thread::sleep_for(std::chrono::seconds(1));
    guard.set_trigger_value(true);
}).detach();

// Wait will return when guard is triggered
waitset.wait();

// Cleanup
waitset.detach(guard);

Infinite Wait​

waitset.wait();  // Blocks until condition triggered

Logging​

HDDS uses standard logging via environment variables. No initialization required.

# Set log level via environment
export RUST_LOG=hdds=info

# Debug level
export RUST_LOG=hdds=debug

# Trace specific modules
export RUST_LOG=hdds::rtps=trace,hdds::discovery=debug

Log levels (via RUST_LOG):

  • error - Errors only
  • warn - Warnings and errors
  • info - Informational messages
  • debug - Debug output
  • trace - Verbose trace output

Telemetry​

Built-in metrics collection.

Initialize​

// Initialize global metrics
hdds::Metrics metrics = hdds::telemetry::init();

Snapshot​

hdds::MetricsSnapshot snap = metrics.snapshot();

std::cout << "Messages sent: " << snap.messages_sent << "\n";
std::cout << "Messages received: " << snap.messages_received << "\n";
std::cout << "Latency P50: " << snap.latency_p50_ms() << "ms\n";
std::cout << "Latency P99: " << snap.latency_p99_ms() << "ms\n";
std::cout << "Latency P99.9: " << snap.latency_p999_ms() << "ms\n";

MetricsSnapshot fields:

struct MetricsSnapshot {
    uint64_t timestamp_ns;
    uint64_t messages_sent;
    uint64_t messages_received;
    uint64_t messages_dropped;
    uint64_t bytes_sent;
    uint64_t latency_p50_ns;
    uint64_t latency_p99_ns;
    uint64_t latency_p999_ns;
    uint64_t merge_full_count;
    uint64_t would_block_count;

    double latency_p50_ms() const;
    double latency_p99_ms() const;
    double latency_p999_ms() const;
};

Exporter (HDDS Viewer)​

// Start telemetry server for HDDS Viewer
auto exporter = hdds::telemetry::start_exporter("127.0.0.1", 4242);

// ... application runs ...

exporter.stop();

Manual Latency Recording​

auto start = std::chrono::steady_clock::now();
// ... operation ...
auto end = std::chrono::steady_clock::now();

metrics.record_latency(
    std::chrono::duration_cast<std::chrono::nanoseconds>(start.time_since_epoch()).count(),
    std::chrono::duration_cast<std::chrono::nanoseconds>(end.time_since_epoch()).count()
);

Error Handling​

// Exceptions
class hdds::Error : public std::runtime_error {
public:
    explicit Error(const std::string& msg);
};

// Example
try {
    auto writer = participant.create_writer_raw("topic");
    writer->write_raw(data);
} catch (const hdds::Error& e) {
    std::cerr << "HDDS error: " << e.what() << "\n";
}

Publisher / Subscriber​

Optional grouping for writers/readers with shared QoS.

// Create publisher
auto publisher = participant.create_publisher(qos);

// Create subscriber
auto subscriber = participant.create_subscriber(qos);

Complete Example​

#include <hdds.hpp>
#include <iostream>
#include <thread>
#include <chrono>

int main() {
    // Logging configured via RUST_LOG environment variable
    // export RUST_LOG=hdds=info

    // Create participant
    hdds::Participant participant("example");

    // Configure QoS
    auto qos = hdds::QoS::reliable()
        .transient_local()
        .history_depth(10);

    // Create writer
    auto writer = participant.create_writer_raw("hello/world", qos);

    // Create reader
    auto reader = participant.create_reader_raw("hello/world", qos);

    // Setup WaitSet
    hdds::WaitSet waitset;
    waitset.attach(reader->get_status_condition());

    // Publisher thread
    std::thread pub_thread([&writer]() {
        for (int i = 0; i < 10; ++i) {
            std::string msg = "Hello #" + std::to_string(i);
            std::vector<uint8_t> data(msg.begin(), msg.end());
            writer->write_raw(data);
            std::cout << "Published: " << msg << "\n";
            std::this_thread::sleep_for(std::chrono::seconds(1));
        }
    });

    // Subscriber loop
    for (int received = 0; received < 10;) {
        if (waitset.wait(std::chrono::seconds(5))) {
            while (auto data = reader->take_raw()) {
                std::string msg(data->begin(), data->end());
                std::cout << "Received: " << msg << "\n";
                ++received;
            }
        }
    }

    pub_thread.join();
    return 0;
}

Thread Safety​

  • Participant creation/destruction: NOT thread-safe
  • Writer/Reader creation: NOT thread-safe
  • writer->write_raw(): Thread-safe (multiple threads can write)
  • reader->take_raw(): NOT thread-safe (use one reader per thread)
  • QoS methods: NOT thread-safe
  • WaitSet: NOT thread-safe

Using with Typed Data​

For typed data, use hdds_gen to generate C++ types with CDR2 serialization:

hddsgen gen cpp Temperature.idl -o temperature.hpp
#include "temperature.hpp"
#include <hdds.hpp>

// Typed writer
auto writer = participant.create_writer<Temperature>("sensors/temp", qos);
writer->write(Temperature{.sensor_id = 1, .value = 23.5f});

// Typed reader
auto reader = participant.create_reader<Temperature>("sensors/temp", qos);
if (auto temp = reader->take<Temperature>()) {
    std::cout << "Sensor " << temp->sensor_id << ": " << temp->value << "C\n";
}

Listeners​

Listeners provide callback-based notification for DDS entity events. This is an alternative to the polling-based WaitSet pattern. Include hdds_listener.hpp for the C++ listener wrappers.

#include <hdds_listener.hpp>

ReaderListener​

Base class for DataReader event callbacks. Override the methods you need; unoverridden methods are no-ops. The listener must outlive the reader it is attached to.

class hdds::ReaderListener {
public:
    virtual ~ReaderListener() = default;

    // Called when new data is available (raw serialized bytes)
    virtual void on_data_available(const uint8_t* data, size_t len);

    // Called when the reader matches/unmatches with a writer
    virtual void on_subscription_matched(const hdds::SubscriptionMatchedStatus& status);

    // Called when liveliness of a matched writer changes
    virtual void on_liveliness_changed(const hdds::LivelinessChangedStatus& status);

    // Called when samples are lost (gap in sequence numbers)
    virtual void on_sample_lost(const hdds::SampleLostStatus& status);

    // Called when samples are rejected due to resource limits
    virtual void on_sample_rejected(const hdds::SampleRejectedStatus& status);

    // Called when the requested deadline is missed
    virtual void on_deadline_missed(const hdds::DeadlineMissedStatus& status);

    // Called when QoS is incompatible with a matched writer
    virtual void on_incompatible_qos(const hdds::IncompatibleQosStatus& status);
};

WriterListener​

Base class for DataWriter event callbacks. Override the methods you need.

class hdds::WriterListener {
public:
    virtual ~WriterListener() = default;

    // Called after a sample is successfully written
    virtual void on_sample_written(const uint8_t* data, size_t len, uint64_t seq);

    // Called when the writer matches/unmatches with a reader
    virtual void on_publication_matched(const hdds::PublicationMatchedStatus& status);

    // Called when an offered deadline is missed
    virtual void on_offered_deadline_missed(uint64_t instance_handle);

    // Called when QoS is incompatible with a matched reader
    virtual void on_offered_incompatible_qos(uint32_t policy_id, const char* policy_name);

    // Called when liveliness is lost (MANUAL_BY_* only)
    virtual void on_liveliness_lost();
};

Status Types​

struct hdds::SubscriptionMatchedStatus {
    uint32_t total_count;           // Total cumulative matched publications
    int32_t  total_count_change;    // Change since last callback
    uint32_t current_count;         // Current number of matched publications
    int32_t  current_count_change;  // Change since last callback
};

struct hdds::PublicationMatchedStatus {
    uint32_t total_count;
    int32_t  total_count_change;
    uint32_t current_count;
    int32_t  current_count_change;
};

struct hdds::LivelinessChangedStatus {
    uint32_t alive_count;
    int32_t  alive_count_change;
    uint32_t not_alive_count;
    int32_t  not_alive_count_change;
};

struct hdds::SampleLostStatus {
    uint32_t total_count;
    int32_t  total_count_change;
};

struct hdds::SampleRejectedStatus {
    uint32_t total_count;
    int32_t  total_count_change;
    uint32_t last_reason;  // 0=NotRejected, 1=ResourceLimit, 2=InstanceLimit, 3=SamplesPerInstanceLimit
};

struct hdds::DeadlineMissedStatus {
    uint32_t total_count;
    int32_t  total_count_change;
};

struct hdds::IncompatibleQosStatus {
    uint32_t total_count;
    int32_t  total_count_change;
    uint32_t last_policy_id;
};

Installing Listeners​

Use hdds::set_listener() to install a listener on a reader or writer, and hdds::clear_listener() to remove it. Both are overloaded for readers and writers.

// Install a ReaderListener
hdds::set_listener(reader->c_handle(), &my_reader_listener);  // returns 0 on success

// Remove a ReaderListener
hdds::clear_listener(reader->c_handle());

// Install a WriterListener
hdds::set_listener(writer->c_handle(), &my_writer_listener);

// Remove a WriterListener
hdds::clear_listener(writer->c_handle());

Per-Callback Convenience Setters​

For simple cases where you only need one callback, use the per-callback setters instead of a full listener class:

// Set a simple on_data_available callback
hdds::set_on_data_available(reader->c_handle(),
    [](const uint8_t* data, size_t len, void* ctx) {
        std::cout << "Received " << len << " bytes\n";
    });

// Set a simple on_subscription_matched callback
hdds::set_on_subscription_matched(reader->c_handle(),
    [](const HddsSubscriptionMatchedStatus* status, void* ctx) {
        std::cout << "Matched writers: " << status->current_count << "\n";
    });

// Set a simple on_publication_matched callback
hdds::set_on_publication_matched(writer->c_handle(),
    [](const HddsPublicationMatchedStatus* status, void* ctx) {
        std::cout << "Matched readers: " << status->current_count << "\n";
    });

Complete Listener Example​

#include <hdds.hpp>
#include <hdds_listener.hpp>
#include <iostream>
#include <atomic>

class SensorListener : public hdds::ReaderListener {
    std::atomic<uint64_t> received_{0};

public:
    void on_data_available(const uint8_t* data, size_t len) override {
        received_.fetch_add(1, std::memory_order_relaxed);
        std::cout << "Sample received (" << len << " bytes)\n";
    }

    void on_subscription_matched(const hdds::SubscriptionMatchedStatus& s) override {
        std::cout << "Matched writers: " << s.current_count
                  << " (total: " << s.total_count << ")\n";
    }

    void on_sample_lost(const hdds::SampleLostStatus& s) override {
        std::cout << "WARNING: " << s.total_count_change << " sample(s) lost\n";
    }

    uint64_t count() const { return received_.load(std::memory_order_relaxed); }
};

int main() {
    hdds::Participant participant("listener_demo");
    auto qos = hdds::QoS::reliable().transient_local();
    auto reader = participant.create_reader_raw("sensors/temp", qos);

    SensorListener listener;
    hdds::set_listener(reader->c_handle(), &listener);

    // ... application runs; callbacks fire on background threads ...

    // Cleanup: remove listener before destroying reader
    hdds::clear_listener(reader->c_handle());
    return 0;
}
Lifetime and Thread Safety
  • The listener object must outlive the reader/writer it is attached to.
  • Callbacks are invoked from background threads. Implementations should return quickly and must not block.
  • Always call hdds::clear_listener() before destroying the listener object.

Not Yet Implemented (v1.0.0)​

FeatureStatus
DSCP C++ bindingsUse env var HDDS_DSCP instead
Instance management (dispose, unregister)Not implemented
SampleInfo with metadataNot implemented
Content-filtered topicsNot implemented

Next Steps​