File dcps.hpp
File List > astutedds > dcps > dcps.hpp
Go to the documentation of this file
#ifndef ASTUTEDDS_DCPS_DCPS_HPP
#define ASTUTEDDS_DCPS_DCPS_HPP
#include <astutedds/dcps/qos.hpp>
#include <astutedds/dcps/dds_psm_types.hpp>
#include <astutedds/dcps/liveliness_manager.hpp>
#include <astutedds/dcps/timer_service.hpp>
#include <astutedds/dcps/topic_description.hpp>
#include <astutedds/rtps/rtps_types.hpp>
#include <astutedds/rtps/udp_transport.hpp>
#include <array>
#include <memory>
#include <mutex>
#include <optional>
#include <vector>
#include <deque>
#include <string>
#include <functional>
#include <set>
#include <map>
#include <unordered_map>
#include <atomic>
#include <thread>
#include <chrono>
#include <condition_variable>
namespace astutedds::dcps
{
// Forward declarations
class DomainParticipant;
class DomainParticipantFactory;
class Topic;
class Publisher;
class Subscriber;
class DataWriter;
class DataReader;
class ContentFilteredTopic;
// ReturnCode_t is now defined as an unscoped enum in dds_psm_types.hpp
// (included above). No duplicate definition here.
struct SampleInfo
{
bool valid_data{false};
rtps::Time_t source_timestamp{};
InstanceHandle_t instance_handle{HANDLE_NIL};
InstanceHandle_t publication_handle{HANDLE_NIL};
int32_t disposed_generation_count{0};
int32_t no_writers_generation_count{0};
int32_t sample_rank{0};
int32_t generation_rank{0};
int32_t absolute_generation_rank{0};
// DDS PSM state fields
SampleStateKind sample_state{NOT_READ_SAMPLE_STATE};
ViewStateKind view_state{NEW_VIEW_STATE};
InstanceStateKind instance_state{ALIVE_INSTANCE_STATE};
};
struct CacheChange
{
rtps::SequenceNumber_t sequenceNumber{};
std::vector<uint8_t> serializedData;
rtps::Time_t sourceTimestamp{};
rtps::ChangeKind_t kind{rtps::ChangeKind_t::ALIVE};
};
struct ReceivedSample
{
rtps::SequenceNumber_t sequenceNumber{};
std::vector<uint8_t> serializedData;
rtps::Time_t sourceTimestamp{};
rtps::GUID_t writerGuid{};
bool valid{true};
bool read{false};
int32_t ownershipStrength{0}; // For EXCLUSIVE ownership filtering
InstanceHandle_t instance_handle{HANDLE_NIL}; // Computed from @key fields
};
class Topic : public TopicDescription
{
public:
Topic(const std::string &name, const std::string &type_name, DomainParticipant *participant);
~Topic() override = default;
// TopicDescription interface
const char *get_name() const override { return name_.c_str(); }
const char *get_type_name() const override { return type_name_.c_str(); }
// Legacy accessors (used by shapes_demo)
const std::string &name() const { return name_; }
const std::string &type_name() const { return type_name_; }
DomainParticipant *participant() const { return participant_; }
void set_qos(const TopicQos &qos) { qos_ = qos; }
const TopicQos &get_qos() const { return qos_; }
private:
std::string name_;
std::string type_name_;
DomainParticipant *participant_;
TopicQos qos_;
};
class DataWriter
{
public:
DataWriter(Topic *topic, Publisher *publisher);
virtual ~DataWriter() = default;
DataWriter(const DataWriter &) = delete;
DataWriter &operator=(const DataWriter &) = delete;
bool write(const std::vector<uint8_t> &data);
bool write(const std::vector<uint8_t> &xcdr1Data,
const std::vector<uint8_t> &xcdr2Data);
uint64_t register_instance();
bool unregister_instance(uint64_t handle);
bool dispose(uint64_t handle);
ReturnCode_t wait_for_acknowledgments(const Duration_t &max_wait)
{
if (qos_.reliability.kind != ReliabilityQosPolicyKind::RELIABLE_RELIABILITY_QOS)
return ReturnCode_t::RETCODE_OK;
// Target: all samples written so far have been acknowledged.
// last_acked_sn_ is advanced by process_acknack() as readers confirm receipt.
const uint64_t target =
next_sequence_number_.low > 1 ? next_sequence_number_.low - 1 : 0;
if (last_acked_sn_ >= target)
return ReturnCode_t::RETCODE_OK;
const auto duration = std::chrono::seconds(max_wait.sec)
+ std::chrono::nanoseconds(max_wait.nanosec);
std::unique_lock<std::mutex> lock(mutex_);
const bool acked = ack_cv_.wait_for(lock, duration,
[this, target] { return last_acked_sn_ >= target; });
return acked ? ReturnCode_t::RETCODE_OK : ReturnCode_t::RETCODE_TIMEOUT;
}
// Accessors
Topic *topic() const { return topic_; }
Topic *get_topic() const { return topic_; } // PSM alias
Publisher *publisher() const { return publisher_; }
const rtps::GUID_t &guid() const { return guid_; }
void set_guid(const rtps::GUID_t &guid) { guid_ = guid; }
void set_qos(const DataWriterQos &qos);
const DataWriterQos &get_qos() const { return qos_; }
const std::deque<CacheChange> &history_cache() const { return history_cache_; }
// Reliability support
void add_matched_reader(const rtps::GUID_t &reader_guid);
void remove_matched_reader(const rtps::GUID_t &reader_guid);
void note_offered_incompatible_qos(QosPolicyId_t failed_policy_id);
void process_acknack(const rtps::GUID_t &reader_guid,
const std::set<uint32_t> &requested_sns,
rtps::Count_t count);
// Transport callback - set by DomainParticipant
using SendCallback = std::function<void(const std::vector<uint8_t> &)>;
void set_send_callback(SendCallback callback) { send_callback_ = callback; }
// Dual-encoding transport callback (XCDR1 + XCDR2)
using SendDualCallback = std::function<void(const std::vector<uint8_t> &,
const std::vector<uint8_t> &)>;
void set_send_dual_callback(SendDualCallback callback) { send_dual_callback_ = callback; }
// Intra-participant loopback callback — invoked from write() alongside
// send_callback_ so a DataReader on the same DomainParticipant receives
// this writer's samples without going through the RTPS/UDP path
// (which filters out self-originated packets). Set by
// Publisher::create_datawriter to route via DomainParticipant::local_deliver().
using LocalDeliverCallback =
std::function<void(const std::vector<uint8_t> &data,
const rtps::GUID_t &writer_guid,
const rtps::SequenceNumber_t &sequence_number)>;
void set_local_deliver_callback(LocalDeliverCallback callback)
{
local_deliver_callback_ = std::move(callback);
}
// Heartbeat callback — invoked by send_heartbeat(); set by DomainParticipant
// to trigger an out-of-band HEARTBEAT submessage via the transport.
using HeartbeatCallback = std::function<void()>;
void set_heartbeat_callback(HeartbeatCallback callback) { heartbeat_callback_ = std::move(callback); }
// DataWriter listener
void set_listener(DataWriterListener *listener, StatusMask mask = STATUS_MASK_ALL)
{
listener_ = listener;
listener_mask_ = mask;
}
DataWriterListener *get_listener() const { return listener_; }
StatusMask get_listener_mask() const { return listener_mask_; }
// Status accessors (DDS §2.2.4.1). Each getter atomically returns the
// current status snapshot and resets the corresponding `*_change` fields
// to zero so subsequent reads only report deltas since the last call.
void get_publication_matched_status(PublicationMatchedStatus &out);
void get_offered_incompatible_qos_status(OfferedIncompatibleQosStatus &out);
LivelinessLostStatus get_liveliness_lost_status() const;
bool assert_liveliness();
using LivelinessAssertHook = std::function<void(const rtps::GUID_t&)>;
void set_liveliness_assert_hook(LivelinessAssertHook hook)
{
liveliness_assert_hook_ = std::move(hook);
}
using LivelinessLostStatusHook = std::function<LivelinessLostStatus()>;
void set_liveliness_lost_hook(LivelinessLostStatusHook hook)
{
liveliness_lost_hook_ = std::move(hook);
}
OfferedDeadlineMissedStatus get_offered_deadline_missed_status();
void set_deadline_timer_service(TimerService *ts)
{
deadline_timer_service_ = ts;
}
// Deadline monitoring (per DDS §2.2.3.15 — scoped to each keyed
// instance). start_deadline_timer() arms a default HANDLE_NIL bucket
// so silent writers still report offered-deadline-missed; per-instance
// buckets are created lazily by notify_write() as new keys are seen.
void start_deadline_timer();
void stop_deadline_timer();
void notify_write(InstanceHandle_t handle); // Reset/arm per-instance timer
// Lifespan: remove expired samples from cache
void remove_expired_samples();
private:
void send_heartbeat();
void retransmit_samples(const std::set<uint32_t> &sequence_numbers);
bool check_resource_limits();
void apply_history_and_limits();
// Fires from the TimerService worker when the deadline period elapses
// without a write() for the given instance. Bumps the offered-
// deadline-missed counters (populating last_instance_handle),
// dispatches the writer / participant listener, then reschedules.
void on_deadline_missed_(InstanceHandle_t handle);
Topic *topic_;
Publisher *publisher_;
rtps::GUID_t guid_;
rtps::SequenceNumber_t next_sequence_number_{0, 1};
std::deque<CacheChange> history_cache_;
DataWriterQos qos_;
mutable std::mutex mutex_;
SendCallback send_callback_;
SendDualCallback send_dual_callback_;
HeartbeatCallback heartbeat_callback_;
LocalDeliverCallback local_deliver_callback_;
// Matched reader tracking (for wait_for_acknowledgments and status reporting)
std::set<rtps::GUID_t> matched_readers_;
// Acknowledgment tracking: last_acked_sn_ is set to the highest SN
// confirmed by all matched readers; ack_cv_ is signaled on each update.
uint64_t last_acked_sn_{0};
std::condition_variable ack_cv_;
// DataWriter listener
DataWriterListener *listener_{nullptr};
StatusMask listener_mask_{STATUS_MASK_ALL};
// Persistent status counters (drained by getters per DDS §2.2.4.1).
PublicationMatchedStatus publication_matched_status_{};
OfferedIncompatibleQosStatus offered_incompatible_qos_status_{};
// §1.1b-B — accumulated across deadline-missed events until read.
OfferedDeadlineMissedStatus offered_deadline_missed_status_{};
// §1.1b-B liveliness hooks — set by Publisher::create_datawriter to
// route assertion and status queries through the participant's
// LivelinessManager. Left null when no participant is attached
// (e.g. unit tests that construct DataWriters directly).
LivelinessAssertHook liveliness_assert_hook_;
LivelinessLostStatusHook liveliness_lost_hook_;
// §1.1b-B deadline watchdog on TimerService — replaces the old
// dedicated deadline_thread_ / cv / running_ trio. The
// TimerService is owned by the participant and shared across
// all writers. §1.1b-B item 10 — one TimerId per keyed
// instance (DDS §2.2.3.15); HANDLE_NIL acts as the default
// bucket for unkeyed topics and for writers armed before any
// write() has produced an instance handle.
TimerService *deadline_timer_service_{nullptr};
std::map<InstanceHandle_t, TimerId> deadline_timers_;
// Lifespan tracking — each cache entry gets a creation timestamp
std::deque<std::chrono::steady_clock::time_point> cache_timestamps_;
};
class DataReader
{
public:
DataReader(Topic *topic, Subscriber *subscriber);
virtual ~DataReader();
DataReader(const DataReader &) = delete;
DataReader &operator=(const DataReader &) = delete;
std::vector<ReceivedSample> read(size_t max_samples = 0xFFFFFFFF);
std::vector<ReceivedSample> take(size_t max_samples = 0xFFFFFFFF);
ReturnCode_t read_next_sample(std::vector<uint8_t> &data, SampleInfo &info);
ReturnCode_t take_next_sample(std::vector<uint8_t> &data, SampleInfo &info);
void add_sample(const ReceivedSample &sample);
ReturnCode_t get_key_value(std::vector<uint8_t> &key_holder, InstanceHandle_t handle);
ReturnCode_t return_loan(std::vector<std::vector<uint8_t>> & /*data_values*/,
std::vector<SampleInfo> & /*sample_infos*/)
{
return ReturnCode_t::RETCODE_OK;
}
// Accessors
Topic *topic() const { return topic_; }
TopicDescription *get_topicdescription() const; // PSM: may be CFT
Subscriber *subscriber() const { return subscriber_; }
const rtps::GUID_t &guid() const { return guid_; }
void set_guid(const rtps::GUID_t &guid) { guid_ = guid; }
void set_qos(const DataReaderQos &qos);
const DataReaderQos &get_qos() const { return qos_; }
size_t unread_count() const;
// Reliability support
void add_matched_writer(const rtps::GUID_t &writer_guid);
void remove_matched_writer(const rtps::GUID_t &writer_guid);
bool is_writer_matched(const rtps::GUID_t &writer_guid) const;
void note_requested_incompatible_qos(QosPolicyId_t failed_policy_id);
void process_heartbeat(const rtps::GUID_t &writer_guid,
const rtps::SequenceNumber_t &first_sn,
const rtps::SequenceNumber_t &last_sn,
rtps::Count_t count,
bool final);
// Data received callback
using DataCallback = std::function<void(const ReceivedSample &)>;
void set_data_callback(DataCallback callback) { data_callback_ = callback; }
// Listener support (DDS §2.1.4.3 — DataReaderListener dispatch)
void set_listener(DataReaderListener* listener) { listener_ = listener; }
DataReaderListener* get_listener() const { return listener_; }
// Status accessors (DDS §2.2.4.1). See DataWriter counterparts for
// semantics; each call drains the `*_change` fields.
void get_subscription_matched_status(SubscriptionMatchedStatus &out);
void get_requested_incompatible_qos_status(RequestedIncompatibleQosStatus &out);
LivelinessChangedStatus get_liveliness_changed_status() const;
using LivelinessChangedStatusHook = std::function<LivelinessChangedStatus()>;
void set_liveliness_changed_hook(LivelinessChangedStatusHook hook)
{
liveliness_changed_hook_ = std::move(hook);
}
// WaitSet notification — external condition variables to signal on data arrival
using NotifyCallback = std::function<void()>;
void add_notify_callback(NotifyCallback cb)
{
std::lock_guard<std::mutex> lock(mutex_);
notify_callbacks_.push_back(std::move(cb));
}
// Destruction-time callback — external owners (e.g. the C API
// eventfd side-map) register a hook that fires from ~DataReader()
// so per-reader resources can be cleaned up deterministically
// even when the reader is deleted through its parent Subscriber
// rather than through the C API delete function. Callbacks are
// invoked in registration order with no lock held.
using DestroyCallback = std::function<void()>;
void add_destroy_callback(DestroyCallback cb)
{
std::lock_guard<std::mutex> lock(mutex_);
destroy_callbacks_.push_back(std::move(cb));
}
// Topic description override (used when created via ContentFilteredTopic)
void set_topicdescription(TopicDescription *td) { topicdescription_ = td; }
RequestedDeadlineMissedStatus get_requested_deadline_missed_status();
SampleLostStatus get_sample_lost_status();
void bump_sample_lost(int32_t count = 1);
void set_deadline_timer_service(TimerService *ts)
{
deadline_timer_service_ = ts;
}
// Deadline monitoring (per DDS §2.2.3.15 — scoped to each keyed
// instance). start_deadline_timer() arms a default HANDLE_NIL bucket
// for readers with no matched samples yet; per-instance buckets are
// created lazily by notify_receive() as new keys arrive.
void start_deadline_timer();
void stop_deadline_timer();
void notify_receive(InstanceHandle_t handle); // Reset/arm per-instance timer
protected:
TopicDescription *topicdescription_{nullptr}; // non-null when created via CFT
private:
void send_acknack(const rtps::GUID_t &writer_guid);
bool check_resource_limits() const;
void apply_history_and_limits();
// §1.1b-B — fires from the TimerService worker when the reader
// has not received a matching sample for the given instance within
// its DEADLINE period. Bumps the requested-deadline-missed counters
// (populating last_instance_handle), dispatches the reader /
// participant listener, then reschedules.
void on_deadline_missed_(InstanceHandle_t handle);
Topic *topic_;
Subscriber *subscriber_;
rtps::GUID_t guid_;
std::deque<ReceivedSample> history_cache_;
DataReaderQos qos_;
mutable std::mutex mutex_;
DataCallback data_callback_;
DataReaderListener* listener_{nullptr};
std::vector<NotifyCallback> notify_callbacks_;
std::vector<DestroyCallback> destroy_callbacks_;
// Persistent status counters (drained by getters per DDS §2.2.4.1).
SubscriptionMatchedStatus subscription_matched_status_{};
RequestedIncompatibleQosStatus requested_incompatible_qos_status_{};
// §1.1b-B — accumulated across deadline-missed events until read.
RequestedDeadlineMissedStatus requested_deadline_missed_status_{};
// §1.1b-B — accumulated across sample-lost events until read.
SampleLostStatus sample_lost_status_{};
// §1.1b-B liveliness hook — set by Subscriber::create_datareader.
// Snapshot-copied from the participant's LivelinessManager on each
// get_liveliness_changed_status() call.
LivelinessChangedStatusHook liveliness_changed_hook_;
// Matched writer tracking (for process_heartbeat and send_acknack)
std::set<rtps::GUID_t> matched_writers_;
// ACKNACK callback — invoked by send_acknack(); set by DomainParticipant
// to deliver an ACKNACK via the transport for the given writer.
using AcknackCallback = std::function<void(const rtps::GUID_t &)>;
AcknackCallback acknack_callback_;
public:
void set_acknack_callback(AcknackCallback cb) { acknack_callback_ = std::move(cb); }
private:
// Time-based filter: track last accepted time per writer
std::map<rtps::GUID_t, std::chrono::steady_clock::time_point> last_accepted_time_;
// Ownership: per-instance tracking of current owner (EXCLUSIVE)
struct InstanceOwnership {
rtps::GUID_t ownerGuid{};
int32_t ownerStrength{0};
};
std::map<int32_t, InstanceOwnership> instance_owners_; // key = instance_handle hash
// Instance registry: maps instance_handle → serialized key bytes (DDS §2.2.2.5.3)
std::map<InstanceHandle_t, std::vector<uint8_t>> instance_key_data_;
// §1.1b-B deadline watchdog on TimerService — replaces the old
// dedicated deadline_thread_ / cv / running_ trio. §1.1b-B item 10 —
// one TimerId per keyed instance (DDS §2.2.3.15 / §2.2.4.1.5
// last_instance_handle); HANDLE_NIL acts as the default bucket for
// unkeyed topics and for readers armed before any sample has arrived.
TimerService *deadline_timer_service_{nullptr};
std::map<InstanceHandle_t, TimerId> deadline_timers_;
};
class Publisher
{
public:
Publisher(DomainParticipant *participant);
~Publisher() = default;
Publisher(const Publisher &) = delete;
Publisher &operator=(const Publisher &) = delete;
DataWriter *create_datawriter(Topic *topic, const DataWriterQos &qos = DataWriterQos{});
DataWriter *create_datawriter(Topic *topic, const DataWriterQos &qos,
DataWriterListener *listener,
StatusMask mask = STATUS_MASK_ALL)
{
DataWriter *writer = create_datawriter(topic, qos);
if (writer && listener)
{
writer->set_listener(listener, mask);
}
return writer;
}
DataWriterQos default_datawriter_qos() const { return DataWriterQos{}; }
ReturnCode_t get_default_datawriter_qos(DataWriterQos &qos) const
{
qos = DataWriterQos{};
return ReturnCode_t::RETCODE_OK;
}
bool delete_datawriter(DataWriter *writer);
ReturnCode_t begin_coherent_changes()
{
coherent_active_ = true;
return ReturnCode_t::RETCODE_OK;
}
ReturnCode_t end_coherent_changes();
bool is_coherent_active() const { return coherent_active_; }
void buffer_coherent_write(DataWriter *writer,
const std::vector<uint8_t> &data);
void buffer_coherent_write_dual(DataWriter *writer,
const std::vector<uint8_t> &xcdr1,
const std::vector<uint8_t> &xcdr2);
DomainParticipant *participant() const { return participant_; }
void set_qos(const PublisherQos &qos) { qos_ = qos; }
const PublisherQos &get_qos() const { return qos_; }
const std::vector<std::unique_ptr<DataWriter>> &writers() const { return writers_; }
std::vector<DataWriter *> get_datawriters() const
{
std::vector<DataWriter *> result;
result.reserve(writers_.size());
for (const auto &w : writers_)
{
result.push_back(w.get());
}
return result;
}
private:
DomainParticipant *participant_;
PublisherQos qos_;
std::vector<std::unique_ptr<DataWriter>> writers_;
uint32_t next_writer_id_{1};
// Coherent changes buffering
bool coherent_active_{false};
struct CoherentEntry {
DataWriter *writer;
std::vector<uint8_t> xcdr1;
std::vector<uint8_t> xcdr2;
bool dual;
};
std::vector<CoherentEntry> coherent_buffer_;
};
class Subscriber
{
public:
Subscriber(DomainParticipant *participant);
~Subscriber() = default;
Subscriber(const Subscriber &) = delete;
Subscriber &operator=(const Subscriber &) = delete;
DataReader *create_datareader(Topic *topic, const DataReaderQos &qos = DataReaderQos{});
DataReader *create_datareader(TopicDescription *td, const DataReaderQos &qos,
DataReaderListener *listener = nullptr,
StatusMask mask = STATUS_MASK_ALL);
DataReaderQos default_datareader_qos() const { return DataReaderQos{}; }
ReturnCode_t get_default_datareader_qos(DataReaderQos &qos) const
{
qos = DataReaderQos{};
return ReturnCode_t::RETCODE_OK;
}
bool delete_datareader(DataReader *reader);
ReturnCode_t begin_access()
{
access_active_ = true;
return ReturnCode_t::RETCODE_OK;
}
ReturnCode_t end_access()
{
access_active_ = false;
return ReturnCode_t::RETCODE_OK;
}
bool is_access_active() const { return access_active_; }
DomainParticipant *participant() const { return participant_; }
void set_qos(const SubscriberQos &qos) { qos_ = qos; }
const SubscriberQos &get_qos() const { return qos_; }
const std::vector<std::unique_ptr<DataReader>> &readers() const { return readers_; }
std::vector<DataReader *> get_datareaders() const
{
std::vector<DataReader *> result;
result.reserve(readers_.size());
for (const auto &r : readers_)
{
result.push_back(r.get());
}
return result;
}
private:
DomainParticipant *participant_;
SubscriberQos qos_;
std::vector<std::unique_ptr<DataReader>> readers_;
uint32_t next_reader_id_{1};
bool access_active_{false};
};
// ── §1.4 Request / Reply (RPC) primitives ─────────────────────────
//
// The DDS-RPC pattern (OMG formal/2017-04-02) transports service calls
// over two correlated DDS topics: `rq/<service>Request` and
// `rr/<service>Reply`. Every payload carries a 24-byte correlation
// header prepended to the serialised user data:
//
// bytes 0..15 16-byte requester GID (identifies the caller)
// bytes 16..23 int64 sequence number (little-endian)
//
// Requesters generate a stable GID at construction and auto-increment
// the sequence number on every `send_request`. Repliers echo the
// caller's GID + sequence number on every `send_reply` so the
// originating requester can correlate the response. Requesters
// filter incoming replies by GID so they do not see replies destined
// for other requesters on the same service.
struct RequesterQos
{
ReliabilityQosPolicy reliability{};
HistoryQosPolicy history{};
DurabilityQosPolicy durability{};
RequesterQos()
{
reliability.kind = ReliabilityQosPolicyKind::RELIABLE_RELIABILITY_QOS;
}
};
using ReplierQos = RequesterQos;
struct TakenRequest
{
std::vector<uint8_t> payload;
std::array<uint8_t, 16> client_gid{};
int64_t sequence_number{0};
};
struct TakenReply
{
std::vector<uint8_t> payload;
int64_t sequence_number{0};
};
class Requester
{
public:
Requester(DomainParticipant *participant,
const std::string &service_name,
const std::string &request_type_name,
const std::string &reply_type_name,
const RequesterQos &qos);
~Requester();
Requester(const Requester &) = delete;
Requester &operator=(const Requester &) = delete;
int64_t send_request(const std::vector<uint8_t> &payload);
std::optional<TakenReply> take_reply();
const std::array<uint8_t, 16> &gid() const { return gid_; }
const std::string &service_name() const { return service_name_; }
DataReader *reply_reader() const { return reader_; }
DataWriter *request_writer() const { return writer_; }
private:
DomainParticipant *participant_;
std::string service_name_;
std::array<uint8_t, 16> gid_{};
std::atomic<int64_t> next_seq_{1};
// Owned indirectly through participant_'s topics_/publishers_/
// subscribers_ vectors. These raw pointers are non-owning.
Topic *req_topic_{nullptr};
Topic *rep_topic_{nullptr};
Publisher *publisher_{nullptr};
Subscriber *subscriber_{nullptr};
DataWriter *writer_{nullptr};
DataReader *reader_{nullptr};
};
class Replier
{
public:
Replier(DomainParticipant *participant,
const std::string &service_name,
const std::string &request_type_name,
const std::string &reply_type_name,
const ReplierQos &qos);
~Replier();
Replier(const Replier &) = delete;
Replier &operator=(const Replier &) = delete;
std::optional<TakenRequest> take_request();
bool send_reply(const std::array<uint8_t, 16> &client_gid,
int64_t sequence_number,
const std::vector<uint8_t> &payload);
const std::string &service_name() const { return service_name_; }
DataReader *request_reader() const { return reader_; }
DataWriter *reply_writer() const { return writer_; }
private:
DomainParticipant *participant_;
std::string service_name_;
Topic *req_topic_{nullptr};
Topic *rep_topic_{nullptr};
Publisher *publisher_{nullptr};
Subscriber *subscriber_{nullptr};
DataWriter *writer_{nullptr};
DataReader *reader_{nullptr};
};
class DomainParticipant
{
public:
DomainParticipant(uint32_t domain_id, uint32_t participant_id);
~DomainParticipant();
DomainParticipant(const DomainParticipant &) = delete;
DomainParticipant &operator=(const DomainParticipant &) = delete;
bool enable();
void stop();
void set_participant_lease_duration_seconds(uint32_t sec);
void set_spdp_announce_interval_ms(uint32_t ms);
// ── Topic creation ──────────────────────────────────────────────
Topic *create_topic(const std::string &topic_name,
const std::string &type_name,
const TopicQos &qos = TopicQos{});
Topic *create_topic(const char *topic_name,
const char *type_name,
const TopicQos &qos,
TopicListener *listener,
StatusMask mask = STATUS_MASK_ALL)
{
(void)listener;
(void)mask;
return create_topic(std::string(topic_name), std::string(type_name), qos);
}
bool delete_topic(Topic *topic);
// ── ContentFilteredTopic ────────────────────────────────────────
ContentFilteredTopic *create_contentfilteredtopic(
const char *name,
Topic *related_topic,
const char *filter_expression,
const StringSeq &expression_parameters);
bool delete_contentfilteredtopic(ContentFilteredTopic *cft);
// ── Publisher / Subscriber creation ─────────────────────────────
Publisher *create_publisher(const PublisherQos &qos = PublisherQos{});
Publisher *create_publisher(const PublisherQos &qos,
PublisherListener *listener,
StatusMask mask = STATUS_MASK_ALL)
{
(void)listener;
(void)mask;
return create_publisher(qos);
}
bool delete_publisher(Publisher *publisher);
Subscriber *create_subscriber(const SubscriberQos &qos = SubscriberQos{});
Subscriber *create_subscriber(const SubscriberQos &qos,
SubscriberListener *listener,
StatusMask mask = STATUS_MASK_ALL)
{
(void)listener;
(void)mask;
return create_subscriber(qos);
}
bool delete_subscriber(Subscriber *subscriber);
// ── §1.4 Request/Reply factories ────────────────────────────────
Requester *create_requester(const std::string &service_name,
const std::string &request_type_name,
const std::string &reply_type_name,
const RequesterQos &qos = RequesterQos{});
bool delete_requester(Requester *requester);
Replier *create_replier(const std::string &service_name,
const std::string &request_type_name,
const std::string &reply_type_name,
const ReplierQos &qos = ReplierQos{});
bool delete_replier(Replier *replier);
// ── Default QoS out-param accessors (PSM) ──────────────────────
ReturnCode_t get_default_topic_qos(TopicQos &qos) const
{
qos = TopicQos{};
return ReturnCode_t::RETCODE_OK;
}
ReturnCode_t get_default_publisher_qos(PublisherQos &qos) const
{
qos = PublisherQos{};
return ReturnCode_t::RETCODE_OK;
}
ReturnCode_t get_default_subscriber_qos(SubscriberQos &qos) const
{
qos = SubscriberQos{};
return ReturnCode_t::RETCODE_OK;
}
// ── Cleanup ────────────────────────────────────────────────────
ReturnCode_t delete_contained_entities();
// ── Type Support Factory ───────────────────────────────────────
using DataWriterFactory = std::function<DataWriter *(Topic *, Publisher *)>;
using DataReaderFactory = std::function<DataReader *(Topic *, Subscriber *)>;
void register_type_factory(const std::string &type_name,
DataWriterFactory wf, DataReaderFactory rf);
DataWriterFactory get_writer_factory(const std::string &type_name) const;
DataReaderFactory get_reader_factory(const std::string &type_name) const;
// ── Accessors ──────────────────────────────────────────────────
uint32_t domain_id() const { return domain_id_; }
uint32_t participant_id() const { return participant_id_; }
const rtps::GuidPrefix_t &guid_prefix() const { return guid_prefix_; }
rtps::RtpsUdpTransport *rtps_transport() const { return transport_.get(); }
void set_qos(const DomainParticipantQos &qos);
const DomainParticipantQos &get_qos() const { return qos_; }
bool is_enabled() const { return running_.load(); }
// Listener management
void set_listener(DomainParticipantListener *listener, StatusMask mask = STATUS_MASK_ALL)
{
listener_ = listener;
listener_mask_ = mask;
}
DomainParticipantListener *get_listener() const { return listener_; }
StatusMask get_listener_mask() const { return listener_mask_; }
void send_data(const std::string &topic_name, const std::vector<uint8_t> &data);
void send_data(const std::string &topic_name,
const std::vector<uint8_t> &xcdr1Data,
const std::vector<uint8_t> &xcdr2Data);
using SendDataCallback = std::function<void(const std::string &topic_name, const std::vector<uint8_t> &data)>;
void set_send_data_callback(SendDataCallback callback) { send_data_callback_ = callback; }
using SubscribeCallback = std::function<void(const std::string &topic_name)>;
void set_subscribe_callback(SubscribeCallback callback) { subscribe_callback_ = callback; }
void register_writer(const std::string &topic_name,
const std::string &type_name,
const std::vector<std::string> &partitions = {},
const DataWriterQos &qos = {});
void register_reader(const std::string &topic_name,
const std::string &type_name,
const std::vector<std::string> &partitions = {},
const DataReaderQos &qos = {});
void unregister_writer(const std::string &topic_name);
void unregister_reader(const std::string &topic_name);
void notify_subscription(const std::string &topic_name);
void deliver_data(const std::string &topic_name, const std::vector<uint8_t> &data,
const rtps::GUID_t &writer_guid = rtps::GUID_t{},
int32_t ownershipStrength = 0,
const rtps::SequenceNumber_t &sequenceNumber = rtps::SequenceNumber_t{});
void local_deliver(const std::string &topic_name,
const std::vector<uint8_t> &data,
const rtps::GUID_t &writer_guid,
const rtps::SequenceNumber_t &sequenceNumber);
void replay_history_to_reader(DataWriter *writer);
using DataReceivedCallback = std::function<void(const std::string &topic, const ReceivedSample &)>;
void set_data_received_callback(DataReceivedCallback callback) { data_received_callback_ = callback; }
void match_reader_to_discovered_writers(DataReader *reader, const std::vector<std::string> &subPartitions);
void match_writer_to_discovered_readers(DataWriter *writer, const std::vector<std::string> &pubPartitions);
std::vector<std::pair<std::string, std::string>> get_discovered_topic_names_and_types() const;
struct DiscoveredPublication
{
rtps::GUID_t endpointGuid;
rtps::GuidPrefix_t participantGuid;
std::string topicName;
std::string typeName;
std::vector<std::string> partitions;
std::vector<uint8_t> participantUserData;
std::string participantName;
ReliabilityQosPolicy reliability;
DurabilityQosPolicy durability;
OwnershipQosPolicy ownership;
int32_t ownershipStrength{0};
DeadlineQosPolicy deadline;
DataRepresentationQosPolicy data_representation;
std::chrono::steady_clock::time_point lastSeen;
};
struct DiscoveredSubscription
{
rtps::GUID_t endpointGuid;
rtps::GuidPrefix_t participantGuid;
std::string topicName;
std::string typeName;
std::vector<std::string> partitions;
std::vector<uint8_t> participantUserData;
std::string participantName;
ReliabilityQosPolicy reliability;
DurabilityQosPolicy durability;
OwnershipQosPolicy ownership;
DeadlineQosPolicy deadline;
DataRepresentationQosPolicy data_representation;
std::chrono::steady_clock::time_point lastSeen;
};
std::vector<DiscoveredPublication> get_discovered_publications() const;
std::vector<DiscoveredSubscription> get_discovered_subscriptions() const;
#ifdef ASTUTEDDS_COMMUNITY_EDITION
// Community Edition endpoint slot management — one slot per DataWriter/DataReader.
bool acquire_writer_slot();
bool acquire_reader_slot();
void release_writer_slot();
void release_reader_slot();
#endif
private:
void discovery_loop();
void generate_guid_prefix();
rtps::GUID_t create_writer_guid(uint32_t writer_id);
rtps::GUID_t create_reader_guid(uint32_t reader_id);
uint32_t domain_id_;
uint32_t participant_id_;
rtps::GuidPrefix_t guid_prefix_;
DomainParticipantQos qos_;
std::atomic<bool> running_{false};
std::thread discovery_thread_;
mutable std::mutex mutex_;
std::vector<std::unique_ptr<Topic>> topics_;
std::vector<std::unique_ptr<Publisher>> publishers_;
std::vector<std::unique_ptr<Subscriber>> subscribers_;
std::vector<std::unique_ptr<ContentFilteredTopic>> cfts_;
std::vector<std::unique_ptr<Requester>> requesters_;
std::vector<std::unique_ptr<Replier>> repliers_;
SendDataCallback send_data_callback_;
SubscribeCallback subscribe_callback_;
DataReceivedCallback data_received_callback_;
// Listener
DomainParticipantListener *listener_{nullptr};
StatusMask listener_mask_{0};
// Type support factories (populated by register_type_factory)
std::map<std::string, DataWriterFactory> writer_factories_;
std::map<std::string, DataReaderFactory> reader_factories_;
// Built-in RTPS transport
std::unique_ptr<rtps::RtpsUdpTransport> transport_;
// §1.1b-B liveliness watchdog — owns a TimerService and tracks
// lease deadlines for local writers (lost) and remote writers
// (changed). Created in enable(), destroyed in stop().
std::unique_ptr<LivelinessManager> liveliness_manager_;
// §1.1b-B deadline watchdog — shared TimerService used by every
// local DataWriter (offered deadline) and DataReader (requested
// deadline). Created in enable(), stopped in stop() *before*
// publishers_/subscribers_ are cleared so no timer callback can
// dereference a torn-down entity.
std::unique_ptr<TimerService> deadline_timers_;
public:
LivelinessManager *liveliness_manager() { return liveliness_manager_.get(); }
TimerService *deadline_timers() { return deadline_timers_.get(); }
private:
// Cached transport tuning values. Applied at enable() time (before
// transport init()) and also live-propagated if the setter is called
// after enable(). See set_participant_lease_duration_seconds and
// set_spdp_announce_interval_ms.
uint32_t pending_lease_duration_sec_{100};
uint32_t pending_spdp_announce_interval_ms_{3000};
// Tracking sets for SEDP callback deduplication and data delivery gating.
// matchedRemoteWriters_ contains GUIDs of remote writers whose QoS is
// compatible with at least one local reader. deliver_data() only delivers
// samples from writers in this set.
std::set<rtps::GUID_t> matchedRemoteWriters_;
std::set<rtps::GUID_t> incompatRemoteWriters_;
std::set<rtps::GUID_t> matchedRemoteReaders_;
std::set<rtps::GUID_t> incompatRemoteReaders_;
#ifdef ASTUTEDDS_COMMUNITY_EDITION
// Community Edition endpoint counters (thread-safe).
std::atomic<uint32_t> ce_writer_count_{0};
std::atomic<uint32_t> ce_reader_count_{0};
#endif
};
class DomainParticipantFactory
{
public:
static DomainParticipantFactory &instance();
static DomainParticipantFactory *get_instance()
{
return &instance();
}
DomainParticipant *create_participant(uint32_t domain_id,
const DomainParticipantQos &qos = DomainParticipantQos{});
DomainParticipant *create_participant(DomainId_t domain_id,
const DomainParticipantQos &qos,
DomainParticipantListener *listener,
StatusMask mask = STATUS_MASK_ALL)
{
auto *dp = create_participant(static_cast<uint32_t>(domain_id), qos);
if (dp && listener)
{
dp->set_listener(listener, mask);
}
return dp;
}
ReturnCode_t get_default_participant_qos(DomainParticipantQos &qos) const
{
qos = DomainParticipantQos{};
return ReturnCode_t::RETCODE_OK;
}
bool delete_participant(DomainParticipant *participant);
private:
DomainParticipantFactory() = default;
std::mutex mutex_;
std::vector<std::unique_ptr<DomainParticipant>> participants_;
std::unordered_map<DomainParticipant*, uint32_t> ref_counts_;
uint32_t next_participant_id_{0};
};
} // namespace astutedds::dcps
// Include Cyclone-style namespace compatibility layer after class definitions
#include <astutedds/dcps/CorePolicy.hpp>
// Standard DDS namespace alias for compatibility
namespace DDS = astutedds::dcps;
#endif // ASTUTEDDS_DCPS_DCPS_HPP