Adds further ext deps as-is binaries copies from customers env.
Qt6 Migration CI / migrate (push) Canceled after 0s

This commit is contained in:
torsten committed 2026-10-23 16:13:31 +00:00
1 parent 48129fdab2
commit 48d9b9bd49
3700 files changed
+2132394

No files matched your search

+53
View File
@@ -0,0 +1,53 @@
#pragma once
#include "ddkit/base/ConfigRw.hpp"
#include "ddkit/base/ConfigLoader.hpp"
#include "ddkit/access/System.hpp"
#include <string>
#include <set>
#include <memory>
#include <utility>
namespace ddkit {
namespace access {
/// @brief ConfigLoaderDds holds the ConfigLoader (Config) and the DdsSystem (process-wide dds-initialization)
/// if created before everything else (on the stack), this will do for clean termination.
class ConfigLoaderDds;
typedef std::shared_ptr<ConfigLoaderDds> ConfigLoaderDdsPtr;
class ConfigLoaderDds {
public:
ConfigLoaderDds();
ConfigLoaderDds(const std::string& configFile);
~ConfigLoaderDds();
base::IConfig& config() { return m_configRw; }
base::IConfigWrite& configWrite() { return m_configRw; }
// checks all keys, which start '<section>.' to be '<section>.<key>' with a <key> from given list
// - checks whole config(-file) right now.
// - enables checking from now on in all calls to above config-functions has(), get(), set().
// the checking-function will throw, if a violation is detected.
void registerKnownConfigKey(const std::string& section, const std::string& key);
void reloadConfig(const std::string& configFile);
static ConfigLoaderDdsPtr instance();
private:
// no copying
ConfigLoaderDds(ConfigLoaderDds const&) = delete;
ConfigLoaderDds(ConfigLoaderDds&&) = delete;
void operator=(ConfigLoaderDds const&) = delete;
void operator=(ConfigLoaderDds&&) = delete;
void loadDefaultTerminateHandlers();
std::unique_ptr<base::ConfigLoader> m_configLoader;
base::ConfigRw m_configRw;
};
} // namespace access
} // namespace ddkit
+47
View File
@@ -0,0 +1,47 @@
#pragma once
#include "ddkit/access/IConnectUpdate.hpp"
#include "ddkit/access/IConnectCheck.hpp"
#include <mutex>
#include <condition_variable>
namespace ddkit {
namespace access {
class ConnectState final : public IConnectUpdate,
public IConnectCheck {
public:
ConnectState(const std::string& name, const std::string& nameModule, const std::string& nameTopic);
// IConnectUpdate
void setIsConnected() override;
void setIsNotConnected() override;
// IConnectCheck
std::string name() override;
std::string nameModule() override;
std::string nameTopic() override;
bool isConnected() override;
void waitUntilConnected();
private:
std::string m_name;
std::string m_nameModule;
std::string m_nameTopic;
std::recursive_mutex m_mutex;
std::condition_variable_any m_condition;
bool m_isConnected;
// no copying
ConnectState(ConnectState const&) = delete;
ConnectState(ConnectState&&) = delete;
void operator=(ConnectState const&) = delete;
};
typedef std::shared_ptr<ConnectState> ConnectStatePtr;
} // namespace access
} // namespace ddkit
+37
View File
@@ -0,0 +1,37 @@
#pragma once
//
// only ever include DdsWrap.hpp instead of dds/dds.hpp!
//
// - visual studio-mechanism for isystem (external includes), that is avoid warnings in third-party-code, is not
// working with (our) cmake so far.
// - so we need to disable all warnings from dds-vendor "by hand". this is done
// - here, while including dds-headers directly
// - in all wrapper-headers around dds-generated files (which include dds indirectly as well obviously)
// - see genVendorWrap for that.
#ifdef _WIN32
#pragma warning(push, 0)
#endif
#include <dds/dds.hpp>
#ifdef _WIN32
#pragma warning(pop)
#endif
// the ostream-operator (operator<<) for InstanceHandle is declared in connext and cyclone nearly the same way
// - connext: ndds/hpp/dds/core/InstanceHandle.hpp
// - cyclone: ddscxx/dds/core/InstanceHandle.hpp
// but connext implements it inline (delegates to rhs.delegate()), while cyclone does not.
// this seems to be the reason, for that the operator is not usable with cyclone at first glance.
// but we simply need to make it known in the namespace for ADL (argument-dependent-lookup) to work.
namespace dds {
namespace core {
using ::operator<<;
}
} // namespace dds
+51
View File
@@ -0,0 +1,51 @@
#pragma once
#include "ddkit/base/Thread.hpp"
#include "ddkit/access/ConfigLoaderDds.hpp"
#include "ddkit/access/DdsWrap.hpp"
#include "ddkit/access/System.hpp"
#include "ddkit/access/ICleanup.hpp"
#include "ddkit/utils/Logger.hpp"
#include <atomic>
namespace ddkit {
namespace access {
class Dispatcher : public base::Thread {
public:
explicit Dispatcher(const std::string& threadName);
Dispatcher(const std::string& threadName, std::shared_ptr<ICleanup> cleanup);
~Dispatcher();
void addReadCondition(const dds::sub::cond::ReadCondition& readCondition);
void removeReadCondition(const dds::sub::cond::ReadCondition& readCondition);
void requestCleanup();
// Thread
void run() override;
void onStop() override;
private:
// System has to be referenced because System::instance() call (and therefore instantiation of the singleton)
// has to happen before the waitset is initialized. Otherwise the waitset would access the policy first and
// the manipulation by System later will throw an exception because the policy is already immutable.
std::shared_ptr<access::ConfigLoaderDds> m_configLoaderDds; // contains System
std::shared_ptr<ICleanup> m_cleanup;
std::recursive_mutex m_addRemoveMutex;
dds::core::cond::WaitSet m_waitset;
dds::core::cond::GuardCondition m_guardCondition;
void callCleanupIf();
Dispatcher(Dispatcher const&) = delete;
void operator=(Dispatcher const&) = delete;
};
} // namespace access
} // namespace ddkit
+17
View File
@@ -0,0 +1,17 @@
#pragma once
namespace ddkit {
namespace access {
class ICleanup {
public:
virtual void cleanup() = 0;
virtual ~ICleanup() = default;
virtual bool isShutdownRequested() = 0;
};
} // namespace access
} // namespace ddkit
+21
View File
@@ -0,0 +1,21 @@
#pragma once
#include <string>
#include <memory>
namespace ddkit {
namespace access {
class IConnectCheck {
public:
virtual std::string name() = 0;
virtual std::string nameModule() = 0;
virtual std::string nameTopic() = 0;
virtual bool isConnected() = 0;
virtual ~IConnectCheck() = default;
};
typedef std::shared_ptr<IConnectCheck> IConnectCheckPtr;
} // namespace access
} // namespace ddkit
+17
View File
@@ -0,0 +1,17 @@
#pragma once
#include <memory>
namespace ddkit {
namespace access {
class IConnectUpdate {
public:
virtual void setIsConnected() = 0;
virtual void setIsNotConnected() = 0;
virtual ~IConnectUpdate() = default;
};
typedef std::shared_ptr<IConnectUpdate> IConnectUpdatePtr;
} // namespace access
} // namespace ddkit
+19
View File
@@ -0,0 +1,19 @@
#pragma once
#include "ddkit/access/QosPattern.hpp"
namespace ddkit {
namespace access {
/// contains one or more QosPatterns and returns them on request by typeName.
class IQosStrategy {
public:
virtual QosPattern selectQosPattern(const std::string& typeName) = 0;
virtual ~IQosStrategy() = default;
};
typedef std::shared_ptr<IQosStrategy> IQosStrategyPtr;
} // namespace access
} // namespace ddkit
+22
View File
@@ -0,0 +1,22 @@
#pragma once
#include <string>
namespace ddkit {
namespace access {
/// @brief template-independent base-class of topics (needed to have a type for generically store a topic-map)
class ITopic {
public:
virtual std::string topicName() = 0;
virtual bool hasReader() = 0;
virtual void unloadReader() = 0;
virtual void closeReader() = 0;
virtual bool hasWriter() = 0;
virtual void closeWriter() = 0;
virtual ~ITopic() = default;
};
} // namespace access
} // namespace ddkit
+28
View File
@@ -0,0 +1,28 @@
#pragma once
#include <string>
#include <memory>
namespace ddkit {
namespace access {
class ITopicStrategy {
public:
virtual std::string get(const std::string& typeName) = 0;
virtual ~ITopicStrategy() = default;
protected:
ITopicStrategy() = default;
private:
// no copying
ITopicStrategy(ITopicStrategy const&) = delete;
ITopicStrategy(ITopicStrategy&&) = delete;
void operator=(ITopicStrategy const&) = delete;
void operator=(ITopicStrategy&&) = delete;
};
typedef std::shared_ptr<ITopicStrategy> ITopicStrategyPtr;
} // namespace access
} // namespace ddkit
+226
View File
@@ -0,0 +1,226 @@
#pragma once
#include "ddkit/access/DdsWrap.hpp"
#include "ddkit/access/ConnectState.hpp"
#include "ddkit/utils/Logger.hpp"
namespace ddkit {
namespace access {
std::string policyName(unsigned policyId);
template <class T>
class ListenerTopic : public dds::topic::TopicListener<T> {
public:
void on_inconsistent_topic(
dds::topic::Topic<T>&, dds::core::status::InconsistentTopicStatus const& status) override {
ddkit::log::warn() << "Inconsistent Topic: "
<< dds::topic::topic_type_name<T>::value()
<< ", count=" << status.total_count()
<< "(" << std::showpos << status.total_count_change() << ")";
}
};
template <class T>
class ListenerRead : public dds::sub::DataReaderListener<T> {
ConnectStatePtr m_connectUpdate;
public:
ListenerRead(ConnectStatePtr connectState): m_connectUpdate(connectState) {
DDKIT_LOG_TRACE << DDKIT_LOG_VAR_VALUE(connectState->nameModule())
<< DDKIT_LOG_VAR_VALUE(connectState->nameTopic());
}
/// @brief We use waitsets and read conditions to get the data more granulary.
/// So we actually dont need this on data here in the dds-listeners.
void on_data_available(dds::sub::DataReader<T>& /* dataReader */) override {
/*DDKIT_LOG_TRACE << std::addressof(dataReader)
<< " topic name: "
<< dataReader.topic_description().name()
<< ", topic type: "
<< dataReader.topic_description().type_name();*/
}
void on_liveliness_changed(
dds::sub::DataReader<T>&, const dds::core::status::LivelinessChangedStatus& status) override {
ddkit::log::debug() << "liveliness changed: "
<< dds::topic::topic_type_name<T>::value() << ", " << status.alive_count() << "/"
<< status.not_alive_count();
}
void on_requested_deadline_missed(
dds::sub::DataReader<T>&, const dds::core::status::RequestedDeadlineMissedStatus&) override {
ddkit::log::warn() << "on_requested_deadline_missed";
}
void on_requested_incompatible_qos(
dds::sub::DataReader<T>& /*reader*/,
const dds::core::status::RequestedIncompatibleQosStatus& status) override {
// When partitions do not match it does not mean that there is a problem.
// It may is intended to direct the dataflow on a logical level.
#if defined(DDS_VENDOR_CONNEXT) || defined(DDS_VENDOR_CYCLONE)
if (status.last_policy_id() != DDS_PARTITION_QOS_POLICY_ID) {
#else
if (true) {
#endif
ddkit::log::warn() << "Reader: Incompatible QoS requested: ("
<< std::to_string(status.last_policy_id()) << "): " << policyName(status.last_policy_id()) << ", "
<< dds::topic::topic_type_name<T>::value() << ", count=" << status.total_count() << "("
<< std::showpos << status.total_count_change() << ")";
}
}
void on_sample_lost(dds::sub::DataReader<T>&, const dds::core::status::SampleLostStatus&) override {
ddkit::log::warn() << "on_sample_lost";
}
void on_sample_rejected(dds::sub::DataReader<T>&, const dds::core::status::SampleRejectedStatus& status) override {
if (status.last_reason() != dds::core::status::SampleRejectedState::not_rejected()) {
ddkit::log::warn() << "on_sample_rejected " << status.last_reason();
}
}
void on_subscription_matched(
dds::sub::DataReader<T>&, const dds::core::status::SubscriptionMatchedStatus& status) override {
ddkit::log::debug() << "Subscription matched: "
<< dds::topic::topic_type_name<T>::value() << ", " << status.current_count();
if (status.current_count()) {
m_connectUpdate->setIsConnected();
} else {
m_connectUpdate->setIsNotConnected();
}
}
virtual ~ListenerRead() {
DDKIT_LOG_TRACE
<< DDKIT_LOG_VAR_VALUE(m_connectUpdate->nameModule())
<< DDKIT_LOG_VAR_VALUE(m_connectUpdate->nameTopic());
}
};
template <class T>
class ListenerWrite : public dds::pub::DataWriterListener<T> {
ConnectStatePtr m_connectUpdate;
public:
ListenerWrite(ConnectStatePtr connectState): m_connectUpdate(connectState) {
DDKIT_LOG_TRACE << DDKIT_LOG_VAR_VALUE(m_connectUpdate->nameModule())
<< DDKIT_LOG_VAR_VALUE(m_connectUpdate->nameTopic());
}
void on_liveliness_lost(dds::pub::DataWriter<T>&, dds::core::status::LivelinessLostStatus const&) override {
ddkit::log::warn() << "liveliness lost";
}
void on_offered_deadline_missed(
dds::pub::DataWriter<T>&, dds::core::status::OfferedDeadlineMissedStatus const&) override {
ddkit::log::warn() << "on_offered_deadline_missed";
}
void on_offered_incompatible_qos(
dds::pub::DataWriter<T>&, dds::core::status::OfferedIncompatibleQosStatus const& status) override {
#if defined(DDS_VENDOR_CONNEXT) || defined(DDS_VENDOR_CYCLONE)
if (status.last_policy_id() != DDS_PARTITION_QOS_POLICY_ID) {
#else
if (true) {
#endif
ddkit::log::warn() << "Writer: Incompatible QoS offered: ("
<< std::to_string(status.last_policy_id()) << "): " << policyName(status.last_policy_id()) << ", "
<< dds::topic::topic_type_name<T>::value() << ", count=" << status.total_count() << "("
<< std::showpos << status.total_count_change() << ")";
}
}
void on_publication_matched(
dds::pub::DataWriter<T>&, dds::core::status::PublicationMatchedStatus const& status) override {
ddkit::log::debug() << "Publication matched: " << DDKIT_LOG_VAR_VALUE(status.current_count());
if (status.current_count()) {
m_connectUpdate->setIsConnected();
} else {
m_connectUpdate->setIsNotConnected();
}
}
#if defined(DDS_VENDOR_CONNEXT)
void on_application_acknowledgment(
dds::pub::DataWriter<T>& /*dataWriter*/,
rti::pub::AcknowledgmentInfo const& /*acknowledgment_info*/) override {
DDKIT_LOG_TRACE;
}
void* on_data_request(dds::pub::DataWriter<T>&, rti::core::Cookie const&) override {
DDKIT_LOG_TRACE;
return nullptr;
}
void on_data_return(dds::pub::DataWriter<T>&, void*, rti::core::Cookie const&) override {
DDKIT_LOG_TRACE;
}
void on_destination_unreachable(
dds::pub::DataWriter<T>&, dds::core::InstanceHandle const&, rti::core::Locator const&) override {
DDKIT_LOG_TRACE;
}
void on_instance_replaced(dds::pub::DataWriter<T>&, dds::core::InstanceHandle const& /*handle*/) override {
DDKIT_LOG_TRACE;
}
void on_reliable_reader_activity_changed(
dds::pub::DataWriter<T>&, const rti::core::status::ReliableReaderActivityChangedStatus&) override {
DDKIT_LOG_TRACE;
}
void on_reliable_writer_cache_changed(
dds::pub::DataWriter<T>&, rti::core::status::ReliableWriterCacheChangedStatus const& status) override {
DDKIT_LOG_TRACE;
ddkit::log::debug() << DDKIT_LOG_VAR_VALUE(status.empty_reliable_writer_cache());
ddkit::log::debug() << DDKIT_LOG_VAR_VALUE(status.full_reliable_writer_cache());
ddkit::log::debug() << DDKIT_LOG_VAR_VALUE(status.low_watermark_reliable_writer_cache());
ddkit::log::debug() << DDKIT_LOG_VAR_VALUE(status.high_watermark_reliable_writer_cache());
ddkit::log::debug() << DDKIT_LOG_VAR_VALUE(status.unacknowledged_sample_count());
ddkit::log::debug() << DDKIT_LOG_VAR_VALUE(status.unacknowledged_sample_count_peak());
}
void on_sample_removed(dds::pub::DataWriter<T>&, rti::core::Cookie const&) override {
DDKIT_LOG_TRACE;
}
void on_service_request_accepted(
dds::pub::DataWriter<T>&, rti::core::status::ServiceRequestAcceptedStatus const&) override {
DDKIT_LOG_TRACE;
}
#elif defined(DDS_VENDOR_CYCLONE)
// other vendors may have other extensions
#elif defined(DDS_VENDOR_OPENSPLICE)
// other vendors may have other extensions
#else
#error "DDS_VENDOR not set"
#endif
virtual ~ListenerWrite() {
DDKIT_LOG_TRACE << DDKIT_LOG_VAR_VALUE(m_connectUpdate->nameModule())
<< DDKIT_LOG_VAR_VALUE(m_connectUpdate->nameTopic());
}
};
} // namespace access
} // namespace ddkit
+89
View File
@@ -0,0 +1,89 @@
#pragma once
#include "ddkit/base/Config.hpp"
#include "ddkit/access/TopicPoolRead.hpp"
#include "ddkit/access/TopicPoolWrite.hpp"
namespace ddkit {
namespace access {
class ModelMonitor final {
public:
ModelMonitor(
base::Config& config,
bool isWrite,
bool isRmodel,
const std::string& header,
int domain,
const IQosStrategyPtr& qosStrategy,
const ITopicStrategyPtr& topicStrategy);
// start current module for addition
void startModule(const std::string& moduleName);
// ... for current module
template <typename T>
void addTopic();
// listen forever
void doRun();
private:
base::Config& m_config;
bool m_isWrite;
bool m_isRmodel;
std::string m_header;
int m_domain;
std::shared_ptr<Dispatcher> m_dispatcher;
Participant m_participant;
TopicPoolWrite m_topicPoolWrite;
TopicPoolRead m_topicPoolRead;
std::string m_moduleCurrent;
int m_ctCurrent;
int m_ctTotal;
std::map<std::string, int> m_module2ct;
};
template <typename T>
void ModelMonitor::addTopic() {
DDKIT_LOG_TRACE;
std::string typeName = dds::topic::topic_type_name<T>::value();
if (m_isRmodel) {
ddkit::utils::string::removePrefix(typeName, "VendorData::");
}
std::string partFirst = typeName.substr(0, typeName.find("::"));
std::string partLast = typeName.substr(typeName.rfind("::") + 2);
std::string moduleName = partFirst;
std::string topicName = partLast;
ddkit::log::trace() << DDKIT_LOG_VAR_VALUE(moduleName);
ddkit::log::trace() << DDKIT_LOG_VAR_VALUE(m_moduleCurrent);
if(moduleName != m_moduleCurrent){
ddkit::log::error() << "ModelMonitor::addTopic: mismatch in module name: \"" << m_moduleCurrent << "\" in class and \"" << moduleName << "\" in topic_type_name.";
return;
}
// bei isRmodel muss das config-format anders werden IService.ConfigAbc ...
std::string key = "module." + moduleName + "." + topicName;
bool isOn = m_config.hasKey(key) && m_config.getBoolValue(key);
ddkit::log::trace() << key;
if (isOn) {
++m_ctCurrent;
if (m_isWrite) {
std::ostringstream deadEnd; deadEnd << "not implemented yet"; // todo
ddkit::log::fatal() << deadEnd.str();
throw std::logic_error(deadEnd.str());
// m_topicPoolWrite.createTopicWrite<T>();
} else {
m_topicPoolRead.createTopicRead<T>(
[=](T const& val) { std::cout << m_header << typeName << ": " << val << std::endl; });
}
}
}
} // namespace access
} // namespace ddkit
+29
View File
@@ -0,0 +1,29 @@
#pragma once
#include "ddkit/access/DdsWrap.hpp"
namespace ddkit {
namespace access {
class Participant {
public:
explicit Participant(int domainId, bool connextExtensibleCompliant=false);
int domain();
dds::domain::DomainParticipant& domainParticipant();
private:
void setConnextXTypesSpecCompliant(bool compliant);
dds::domain::qos::DomainParticipantQos getParticipantQos();
bool m_connextExtensibleCompliant;
int m_domainId;
dds::domain::DomainParticipant m_domainParticipant;
// default copying
};
} // namespace access
} // namespace ddkit
+54
View File
@@ -0,0 +1,54 @@
#pragma once
#include "ddkit/access/DdsWrap.hpp"
namespace ddkit {
namespace access {
/// @brief wraps dds::pub::qos::DataWriterQos and dds::sub::qos::DataReaderQos
/// allows defining an arbitrary list of qos-settings.
class QosPattern {
public:
QosPattern();
QosPattern get();
dds::sub::qos::DataReaderQos& qosRead();
dds::pub::qos::DataWriterQos& qosWrite();
dds::topic::qos::TopicQos& qosTopic();
dds::sub::qos::SubscriberQos& qosSub();
dds::pub::qos::PublisherQos& qosPub();
QosPattern& qosReadWrite();
template <typename T>
QosPattern& operator<<(T const& rhs) {
qosRead() << rhs;
qosWrite() << rhs;
qosTopic() << rhs;
return *this;
}
private:
dds::sub::qos::DataReaderQos m_dataReaderQos;
dds::pub::qos::DataWriterQos m_dataWriterQos;
dds::topic::qos::TopicQos m_topicQos;
dds::sub::qos::SubscriberQos m_subscriberQos;
dds::pub::qos::PublisherQos m_publisherQos;
};
class QosPatternDefault : public QosPattern {
public:
QosPatternDefault() {
using namespace dds::core::policy;
qosReadWrite() << Ownership(OwnershipKind::SHARED)
<< Reliability(ReliabilityKind::RELIABLE)
<< Durability(DurabilityKind::TRANSIENT_LOCAL);
}
};
} // namespace access
} // namespace ddkit
+33
View File
@@ -0,0 +1,33 @@
#pragma once
#include "ddkit/access/QosPattern.hpp"
namespace ddkit {
namespace access {
/// @brief Tmsif (rmodel-generated-models/appls)
class QosPatternRm : public QosPattern {
public:
QosPatternRm(const std::vector<std::string>& partitions, const bool ignoreLocal);
};
/// @brief Config/Report
class QosPatternRmLateJoin : public QosPatternRm {
public:
QosPatternRmLateJoin(const std::vector<std::string>& partitions, const bool ignoreLocal);
};
/// @brief Notify/Event / Cmd/Request/Response
class QosPatternRmFireForget : public QosPatternRm {
public:
QosPatternRmFireForget(const std::vector<std::string>& partitions, const bool ignoreLocal);
};
} // namespace access
} // namespace ddkit
+19
View File
@@ -0,0 +1,19 @@
#pragma once
#include "ddkit/access/IQosStrategy.hpp"
#include "ddkit/access/QosPattern.hpp"
namespace ddkit {
namespace access {
// default and example
class QosStrategyDefault : public IQosStrategy {
public:
QosPattern selectQosPattern(const std::string& /*typeName*/) override {
// typeName ignored in this case, set default for all typenames
return QosPatternDefault().get();
}
};
} // namespace access
} // namespace ddkit
+22
View File
@@ -0,0 +1,22 @@
#pragma once
#include "ddkit/access/IQosStrategy.hpp"
#include "ddkit/access/QosPattern.hpp"
namespace ddkit {
namespace access {
class QosStrategyRm : public IQosStrategy {
public:
QosStrategyRm(const std::vector<std::string>& partitions, const bool ignoreLocal);
QosPattern selectQosPattern(const std::string& typeName) override;
private:
std::vector<std::string> m_partitions;
bool m_ignoreLocal = false;
};
} // namespace access
} // namespace ddkit
+171
View File
@@ -0,0 +1,171 @@
#pragma once
#include "ddkit/access/DdsWrap.hpp"
#include "ddkit/access/Dispatcher.hpp"
#include "ddkit/access/Topic.hpp"
#include "ddkit/access/Listener.hpp"
#include "ddkit/access/ConnectState.hpp"
namespace ddkit {
namespace access {
#if defined(DDS_VENDOR_CONNEXT)
#define DDKIT_USE_KEY_VALUE ON
#ifdef _WIN32
#define DDKIT_COND_TYPE_UNUSED
#else
#define DDKIT_COND_TYPE_UNUSED dds::core::cond::Condition
#endif
#elif defined(DDS_VENDOR_CYCLONE)
#define DDKIT_COND_TYPE_UNUSED dds::core::cond::Condition
#elif defined(DDS_VENDOR_OPENSPLICE)
#define DDKIT_COND_TYPE_UNUSED dds::core::cond::Condition
#else
#error "DDS_VENDOR not set"
#endif
template <typename T>
class Reader final {
public:
Reader(
std::shared_ptr<access::Dispatcher> dispatcher,
std::shared_ptr<Topic<T>> topic,
dds::sub::qos::DataReaderQos dataReaderQos,
dds::sub::qos::SubscriberQos subQos,
std::function<void(T const&)> funcWrite,
std::function<void(T const&)> funcDispose)
: m_dispatcher(dispatcher)
, m_connectState(std::make_shared<ConnectState>(topic->topicName(), topic->topicNameModule(), topic->topicNameTopic()))
, m_listener(std::make_shared<ListenerRead<T> >(m_connectState))
, m_ddsReader(dataReader(topic, m_listener.get(), dataReaderQos, subQos))
, m_dataState(dds::sub::status::DataState::any())
, m_shutdownRequest(false)
, m_funcWrite(funcWrite)
, m_funcDispose(funcDispose)
, m_func([this](DDKIT_COND_TYPE_UNUSED) {
DDKIT_LOG_TRACE;
const dds::sub::LoanedSamples<T> samples = m_ddsReader.select().state(m_dataState).take();
for (const auto& sample : samples) {
if (m_shutdownRequest) return; // fast exit
const auto& sampleInfo = sample.info();
#ifdef DDKIT_USE_KEY_VALUE
const auto& instanceHandle = sampleInfo.instance_handle();
#endif
const auto& dataState = sampleInfo.state();
const auto& instanceState = dataState.instance_state();
using namespace dds::sub::status;
ddkit::log::trace() << "Sample alive:" << (instanceState == InstanceState::alive());
ddkit::log::trace() << "Sample disposed:" << (instanceState == InstanceState::not_alive_disposed());
ddkit::log::trace() << "Sample no_writers:" << (instanceState == InstanceState::not_alive_no_writers());
// How to get partition of sending writer
// only connext and cyclone > 0.10
// auto pubData = dds::sub::matched_publication_data(m_ddsReader, sampleInfo.publication_handle());
// for (auto& n : pubData.partition()->name()) {
// std::cout << "Partition: " << n << std::endl;
// }
if (sampleInfo.valid() && instanceState == InstanceState::alive()) {
const T& t = sample.data();
// callback
if (m_funcWrite) {
ddkit::log::debug() << "received: " << t << ". Callbacking user ...";
m_funcWrite(t);
ddkit::log::trace() << "... usercallback called.";
}
} else {
if (m_funcDispose) {
#ifdef DDKIT_USE_KEY_VALUE
// - 'key_value()' works perfectly for connext
// (and for cyclone+opensplice while explicitly disposing)
// - no (positive) effect on cross-vendor (see top of file)
// - so it is best to stay with native sample.data() for cyclone+opensplice but
// use simple key_value for connext
T t;
m_ddsReader.key_value(t, instanceHandle);
#else
// - vendors, that are able to fetch the key even on dispose-calls via sample.data()
// (they fill in default-ctor-data in the other attrs, because only the key has
// to be transferred via DDS for dispose - same definition on our API - fine).
const T& t = sample.data();
#endif
ddkit::log::debug() << "received for disposing: " << t << ". Callbacking user ...";
m_funcDispose(t);
ddkit::log::trace() << "... usercallback called.";
}
}
}
})
, m_readCondition(m_ddsReader, m_dataState, m_func) {
DDKIT_LOG_TRACE;
m_dispatcher->addReadCondition(m_readCondition);
}
~Reader() {
DDKIT_LOG_TRACE;
unload();
m_ddsReader.close();
}
void unload() {
if (!m_shutdownRequest) {
m_shutdownRequest = true;
m_dispatcher->removeReadCondition(m_readCondition);
}
}
IConnectCheckPtr getConnectCheck() {
return m_connectState;
}
private:
std::shared_ptr<Dispatcher> m_dispatcher;
ConnectStatePtr m_connectState;
std::shared_ptr<ListenerRead<T>> m_listener;
dds::sub::DataReader<T> m_ddsReader;
dds::sub::status::DataState m_dataState;
std::atomic<bool> m_shutdownRequest;
std::function<void(T const&)> m_funcWrite;
std::function<void(T const&)> m_funcDispose;
std::function<void(DDKIT_COND_TYPE_UNUSED)> m_func;
dds::sub::cond::ReadCondition m_readCondition;
/// Note on reader-listeners: see writer-listeners in Writer.hpp
static dds::sub::DataReader<T> dataReader(
std::shared_ptr<Topic<T>> topic,
ListenerRead<T>* listener,
dds::sub::qos::DataReaderQos dataReaderQos,
dds::sub::qos::SubscriberQos subQos) {
DDKIT_LOG_TRACE;
dds::sub::DataReader<T> ddsReader = dds::sub::DataReader<T>(
dds::sub::Subscriber(topic->domainParticipant(), subQos),
topic->ddsTopic(),
dataReaderQos,
listener,
dds::core::status::StatusMask::all());
return ddsReader;
}
// no copying
Reader(Reader const&) = delete;
Reader(Reader&&) = delete;
void operator=(Reader const&) = delete;
};
#undef DDKIT_USE_KEY_VALUE
#undef DDKIT_COND_TYPE_UNUSED
} // namespace access
} // namespace ddkit
+18
View File
@@ -0,0 +1,18 @@
#pragma once
namespace ddkit {
namespace access {
class System {
public:
static System& instance();
private:
System();
~System();
};
} // namespace access
} // namespace ddkit
+141
View File
@@ -0,0 +1,141 @@
#pragma once
#include "ddkit/access/DdsWrap.hpp"
#include "ddkit/access/ITopic.hpp"
#include "ddkit/access/Participant.hpp"
#include "ddkit/access/Listener.hpp"
#include "ddkit/utils/Logger.hpp"
namespace ddkit {
namespace access {
template <typename T>
class Reader;
template <typename T>
using ReaderPtr = std::shared_ptr<Reader<T>>;
template <typename T>
class Writer;
template <typename T>
using WriterPtr = std::shared_ptr<Writer<T>>;
template <typename T>
class Topic final : public ITopic {
public:
dds::domain::DomainParticipant& domainParticipant() { return m_domainParticipant; }
dds::topic::Topic<T> ddsTopic() { return m_ddsTopic; }
std::string typeName() { return dds::topic::topic_type_name<T>::value(); }
std::string topicNameModule() { return m_topicNameModule; }
std::string topicNameTopic() { return m_topicNameTopic; }
// ITopic
std::string topicName() override { return m_topicName; }
bool hasReader() override {
std::unique_lock<std::recursive_mutex> lock(m_mutex);
return m_reader != nullptr;
}
void unloadReader() override {
std::unique_lock<std::recursive_mutex> lock(m_mutex);
DDKIT_LOG_TRACE << typeName();
if (hasReader()) {
m_reader->unload();
}
}
void closeReader() override {
std::unique_lock<std::recursive_mutex> lock(m_mutex);
ddkit::log::debug() << "closeReader (" << typeName() << ")";
if (hasReader()) {
m_reader = nullptr;
}
}
bool hasWriter() override {
std::unique_lock<std::recursive_mutex> lock(m_mutex);
return m_writer != nullptr;
}
void closeWriter() override {
std::unique_lock<std::recursive_mutex> lock(m_mutex);
ddkit::log::debug() << "closeWriter (" << typeName() << ")";
if (hasWriter()) {
m_writer = nullptr;
}
}
Topic(
dds::domain::DomainParticipant& domainParticipant,
const std::string& topicName,
const std::string& topicNameModule,
const std::string& topicNameTopic,
dds::topic::qos::TopicQos topicQos) try
: m_domainParticipant(domainParticipant)
, m_topicName(topicName)
, m_topicNameModule(topicNameModule)
, m_topicNameTopic(topicNameTopic)
, m_listener(std::make_shared<ListenerTopic<T> >())
, m_ddsTopic(dds::core::null)
, m_reader(nullptr)
, m_writer(nullptr) {
#if defined(DDS_VENDOR_CONNEXT)
m_ddsTopic = dds::topic::find<dds::topic::Topic<T>>(domainParticipant, topicName);
#endif
if (m_ddsTopic == dds::core::null) {
m_ddsTopic = dds::topic::Topic<T>(
m_domainParticipant,
m_topicName,
topicQos,
m_listener.get(), // has to be set in constructor of topic
dds::core::status::StatusMask::all()
);
}
ddkit::log::info() << "Created topic: " << m_topicName;
} catch (dds::core::Exception& e) {
// happens once in a while on connext (if the same topic is ctored twice in same participant)
ddkit::log::fatal() << "construction Topic " << topicName << " failed: " << e.what();
throw std::runtime_error(e.what());
}
void setReader(ReaderPtr<T> reader) {
std::unique_lock<std::recursive_mutex> lock(m_mutex);
m_reader = reader;
}
std::shared_ptr<Writer<T>> writer() {
std::unique_lock<std::recursive_mutex> lock(m_mutex);
return m_writer;
}
void setWriter(WriterPtr<T> writer) {
std::unique_lock<std::recursive_mutex> lock(m_mutex);
m_writer = writer;
}
private:
bool m_isRead;
dds::domain::DomainParticipant& m_domainParticipant;
std::string m_topicName;
std::string m_topicNameModule;
std::string m_topicNameTopic;
std::shared_ptr<ListenerTopic<T>> m_listener;
dds::topic::Topic<T> m_ddsTopic;
std::recursive_mutex m_mutex;
std::shared_ptr<Reader<T>> m_reader;
std::shared_ptr<Writer<T>> m_writer;
// no copying
Topic(Topic const&) = delete;
Topic(Topic&&) = delete;
void operator=(Topic const&) = delete;
};
} // namespace access
} // namespace ddkit
+100
View File
@@ -0,0 +1,100 @@
#pragma once
#include "ddkit/access/ITopic.hpp"
#include "ddkit/access/IQosStrategy.hpp"
#include "ddkit/access/TopicStrategy.hpp"
#include "ddkit/access/Topic.hpp"
// Topic is included and only forward declares Reader,Writer - for writing-only-users we additionaly need the read-side
// for template-instantiiation ... so include everything just in case ...
#include "ddkit/access/Reader.hpp"
#include "ddkit/access/Writer.hpp"
#include "ddkit/utils/string.hpp"
namespace ddkit {
namespace access {
// TopicPool very well could be the (abstract) base-class for TopicPoolRead and TopicPoolWrite again
class TopicPool final {
public:
TopicPool(
dds::domain::DomainParticipant& domainParticipant,
const IQosStrategyPtr& qosStrategy,
const ITopicStrategyPtr& topicStrategy);
template <typename T>
std::string getTopicName() {
DDKIT_LOG_TRACE;
std::string className = dds::topic::topic_type_name<T>::value();
ddkit::log::trace() << DDKIT_LOG_VAR_VALUE(className);
std::string topicName = m_topicStrategy->get(className);
ddkit::log::trace() << DDKIT_LOG_VAR_VALUE(topicName);
TopicStrategyMapChars topicStrategyMapChars;
std::string topicNameSafe = topicStrategyMapChars.get(topicName);
ddkit::log::trace() << DDKIT_LOG_VAR_VALUE(topicNameSafe);
return topicNameSafe;
}
/// @brief create or get
template <typename T>
std::shared_ptr<Topic<T>> topic() {
DDKIT_LOG_TRACE;
std::string topicName = getTopicName<T>();
if (!hasTopic(topicName)) {
std::string className = dds::topic::topic_type_name<T>::value();
ddkit::utils::string::removePrefixIf(className, "VendorData::");
std::string::size_type pos = className.rfind("::");
std::string partFirst = (pos == std::string::npos) ? "no-module" : className.substr(0, pos);
std::string partLast = (pos == std::string::npos) ? className : className.substr(pos + 2);
std::string topicNameModule = partFirst;
std::string topicNameTopic = partLast;
m_name2topic.insert(std::make_pair(topicName, std::make_shared<Topic<T>>(
m_domainParticipant,
topicName,
topicNameModule,
topicNameTopic,
m_qosStrategy->selectQosPattern(className).qosTopic()
)));
}
return std::dynamic_pointer_cast<Topic<T>>(getTopic(topicName));
}
QosPattern qosPattern(const std::string& typeName) { return m_qosStrategy->selectQosPattern(typeName); }
template <typename T> // probably not needed with template-param
QosPattern qosPattern() {
std::string typeName = dds::topic::topic_type_name<T>::value();
return qosPattern(typeName);
}
bool hasTopic(const std::string& topicName);
std::shared_ptr<ITopic> getTopic(const std::string& topicName);
bool hasReaders();
void closeReaders();
void closeWriters();
void unloadReaders();
private:
dds::domain::DomainParticipant& m_domainParticipant;
IQosStrategyPtr m_qosStrategy;
ITopicStrategyPtr m_topicStrategy;
std::recursive_mutex m_mutex;
std::map<std::string, std::shared_ptr<ITopic>> m_name2topic;
// no copying
TopicPool(TopicPool const&) = delete;
TopicPool(TopicPool&&) = delete;
void operator=(TopicPool const&) = delete;
};
typedef std::shared_ptr<TopicPool> TopicPoolPtr;
} // namespace access
} // namespace ddkit
+78
View File
@@ -0,0 +1,78 @@
#pragma once
#include "ddkit/access/TopicPool.hpp"
#include "ddkit/access/Topic.hpp"
#include "ddkit/access/Dispatcher.hpp"
#include "ddkit/access/Reader.hpp"
#include "ddkit/utils/string.hpp"
#include <utility>
namespace ddkit {
namespace access {
class TopicPoolRead final {
public:
TopicPoolRead(
Participant& participant,
const IQosStrategyPtr& qosStrategy,
const ITopicStrategyPtr& topicStrategy,
std::shared_ptr<Dispatcher> dispatcher);
~TopicPoolRead();
// create Topic and Reader and register function for beeing callbacked by given dispatcher
// (a second callback may be given for reading disposes as well)
// todo name is historical, could be createReader
template <typename T>
void createTopicRead(std::function<void(T const&)> func, std::function<void(T const&)> funcDispose = nullptr) {
DDKIT_LOG_TRACE;
std::unique_lock<std::recursive_mutex> lock(m_mutex);
std::string topicName = m_topicPool.getTopicName<T>();
std::shared_ptr<Topic<T>> topic = m_topicPool.topic<T>();
if(!topic) {
ddkit::log::error() << "TopicPoolRead::createTopicRead: no topic in topic pool.";
return;
}
if(topic->hasReader()) {
ddkit::log::error() << "TopicPoolRead::createTopicRead: Reader (" << topicName << ") already exists.";
return;
}
QosPattern qosPattern = m_topicPool.qosPattern(topic->typeName());
dds::sub::qos::DataReaderQos dataReaderQos = qosPattern.qosRead();
dds::sub::qos::SubscriberQos subQos = qosPattern.qosSub();
auto reader = std::make_shared<Reader<T>>(m_dispatcher, topic, dataReaderQos, subQos, func, funcDispose);
topic->setReader(reader);
}
// to be able to call with member-function-callbacks as well - transform to std::function
// - this is used in ldm/test-context so far, no dispose-callback used here so far ...
template <typename T, typename U>
void createTopicRead(void (U::*funcPtr)(T const&), U& u) {
DDKIT_LOG_TRACE;
createTopicRead<T>(std::bind(funcPtr, &u, std::placeholders::_1));
}
bool hasReaders();
void closeReaders();
void unloadReaders();
private:
std::recursive_mutex m_mutex;
TopicPool m_topicPool;
std::shared_ptr<Dispatcher> m_dispatcher;
// no copying
TopicPoolRead(TopicPoolRead const&) = delete;
TopicPoolRead(TopicPoolRead&&) = delete;
void operator=(TopicPoolRead const&) = delete;
};
} // namespace access
} // namespace ddkit
+137
View File
@@ -0,0 +1,137 @@
#pragma once
#include "ddkit/access/TopicPool.hpp"
#include "ddkit/access/Topic.hpp"
#include "ddkit/access/Writer.hpp"
#include "ddkit/utils/system.hpp"
namespace ddkit {
namespace access {
class TopicPoolWrite final {
public:
TopicPoolWrite(
Participant& participant, const IQosStrategyPtr& qosStrategy, const ITopicStrategyPtr& topicStrategy);
~TopicPoolWrite();
// create or get writer
template <typename T>
std::shared_ptr<Writer<T>> writer() {
DDKIT_LOG_TRACE;
std::unique_lock<std::recursive_mutex> lock(m_mutex);
createTopicWriteIf<T>(); // must be done here and not in Topic (because TopicPool and qos access needed)
std::shared_ptr<Topic<T>> topic = m_topicPool.topic<T>();
if(topic) {
auto writer = topic->writer();
if(!writer){
ddkit::log::error() << "TopicPoolWrite::writer: no writer in topic";
}
return writer;
} else {
ddkit::log::error() << "TopicPoolWrite::writer: no topic in topic pool";
}
return nullptr;
}
// shortcut, no interface+virtual possible (templates)
template <class T>
void write(T const& t) {
bool isOkUnconnected = false;
bool isThrowUnconnected = false;
writeIntern(t, isOkUnconnected, isThrowUnconnected);
}
template <class T>
void writeUnconnectedOk(T const& t) {
bool isOkUnconnected = true;
bool isThrowUnconnected = false;
writeIntern(t, isOkUnconnected, isThrowUnconnected);
}
template <class T>
void writeUnconnectedThrow(T const& t) {
bool isOkUnconnected = false;
bool isThrowUnconnected = false;
writeIntern(t, isOkUnconnected, isThrowUnconnected);
}
private:
template <typename T>
void writeIntern(T const& t, bool isOkUnconnected, bool isThrowUnconnected) {
ddkit::log::trace() << "writing " << dds::topic::topic_type_name<T>::value() << ":" << t;
createTopicWriteIf<T>();
std::unique_lock<std::recursive_mutex> lock(m_mutex);
auto currentWriter = writer<T>();
if(currentWriter){
currentWriter->send(t, isOkUnconnected, isThrowUnconnected);
} else {
ddkit::log::error() << "TopicPoolWrite::writeIntern: no writer in topic pool";
}
}
template <typename T>
void createTopicWrite() {
std::unique_lock<std::recursive_mutex> lock(m_mutex);
std::string topicName = m_topicPool.getTopicName<T>();
std::shared_ptr<Topic<T>> topic = m_topicPool.topic<T>();
std::string typeName = topic->typeName();
if(!topic) {
ddkit::log::error() << "TopicPoolWrite::createTopicWrite: no topic in topic pool.";
return;
}
if(topic->hasWriter()) {
ddkit::log::error() << "TopicPoolWrite::createTopicWrite: Writer (" << topicName << ") already exists.";
return;
}
QosPattern qosPattern = m_topicPool.qosPattern(typeName);
dds::pub::qos::DataWriterQos dataWriterQos = qosPattern.qosWrite();
dds::pub::qos::PublisherQos pubQos = qosPattern.qosPub();
auto writer = std::make_shared<Writer<T>>(topic, dataWriterQos, pubQos);
topic->setWriter(writer);
}
template <class T>
void createTopicWriteIf() {
std::unique_lock<std::recursive_mutex> lock(m_mutex);
std::shared_ptr<Topic<T>> topic = m_topicPool.topic<T>();
if (topic && !topic->hasWriter()) {
createTopicWrite<T>();
}
}
public:
template <class T>
void dispose(T const& t) {
std::unique_lock<std::recursive_mutex> lock(m_mutex);
auto currentWriter = writer<T>();
if(currentWriter){
currentWriter->dispose(t);
} else {
ddkit::log::error() << "TopicPoolWrite::dispose: no writer in topic pool";
}
}
void closeWriters();
private:
std::recursive_mutex m_mutex;
TopicPool m_topicPool;
TopicPoolWrite(TopicPoolWrite const&) = delete;
void operator=(TopicPoolWrite const&) = delete;
};
} // namespace access
} // namespace ddkit
+80
View File
@@ -0,0 +1,80 @@
#pragma once
#include "ddkit/access/ITopicStrategy.hpp"
#include <string>
namespace ddkit {
namespace access {
class TopicStrategyRemoveNamespaces : public ITopicStrategy {
public:
std::string get(const std::string& typeName) override;
};
/// @brief replace "::" with "__" (as for example cyclone does not like '::' in topic-names)
class TopicStrategyReplaceColons : public ITopicStrategy {
public:
std::string get(const std::string& typeName) override;
};
/// @brief simply prepend the prefix (plus "_") - nothing else
class TopicStrategyPrefix : public ITopicStrategy {
public:
TopicStrategyPrefix(const std::string& prefix);
std::string get(const std::string& typeName) override;
private:
std::string m_prefix;
};
/// @brief maps all 'unallowed' characters (everything besides [a-zA-Z0-9]) to '_' - implicitly called in TopicPool - always!
class TopicStrategyMapChars : public ITopicStrategy {
public:
TopicStrategyMapChars();
std::string get(const std::string& orig) override;
};
/// @brief SensorService + VendorData::M_Sensor::IService::ConfigSth -> ddkit_SensorService_Sensor_c2s_ConfigSth
class TopicStrategyRmodel : public ITopicStrategy {
public:
TopicStrategyRmodel(const std::string& prefix);
std::string get(const std::string& typeName) override;
private:
TopicStrategyReplaceColons m_tnsReplaceColons;
TopicStrategyPrefix m_tnsPrefix;
};
/// @brief module::name -> module_name
class TopicStrategySimple : public ITopicStrategy {
public:
std::string get(const std::string& typeName) override;
};
/// @brief remove big-char-prefixes and -suffixes and replace colons by underscore
/// - P_Navigation_Reference_PSM::C_Angular_Motion --> Navigation_Reference__Angular_Motion
class TopicStrategyStd : public ITopicStrategy {
public:
std::string get(const std::string& typeName) override;
};
} // namespace access
} // namespace ddkit
+117
View File
@@ -0,0 +1,117 @@
#pragma once
#include "ddkit/access/DdsWrap.hpp"
#include "ddkit/access/Topic.hpp"
#include "ddkit/access/Listener.hpp"
#include "ddkit/access/ConnectState.hpp"
#include "ddkit/utils/Logger.hpp"
namespace ddkit {
namespace access {
template <typename T>
class Writer final {
public:
explicit Writer(std::shared_ptr<Topic<T>> topic, dds::pub::qos::DataWriterQos dataWriterQos, dds::pub::qos::PublisherQos pubQos)
: m_connectState(std::make_shared<ConnectState>(topic->topicName(), topic->topicNameModule(), topic->topicNameTopic()))
, m_listener(std::make_shared<ListenerWrite<T> >(m_connectState))
, m_writer(dataWriter(topic, m_listener.get(), dataWriterQos, pubQos)) {
DDKIT_LOG_TRACE;
}
~Writer() {
DDKIT_LOG_TRACE;
// ddkit::log::info() << "closing writer!";
m_writer.close();
}
bool isConnected() { return m_connectState->isConnected(); }
void waitUntilConnected() { m_connectState->waitUntilConnected(); }
IConnectCheckPtr getConnectCheck() { return m_connectState; }
/// @brief isOkUnconnected == true -> do not warn about write while not connected
/// - one could set this on a late-join-topic to avoid the warnings
/// isThrowUnconnected == true -> do throw momentarily
/// - one could set this on a 'checked' write ('Cmd')
void send(T const& sample, bool isOkUnconnected, bool isThrowUnconnected) {
ddkit::log::debug() << "send: " << DDKIT_LOG_VAR_VALUE(sample);
bool isConnected = m_connectState->isConnected();
// return immediately, if Service for command is not yet connected (and told to throw)
if (!isConnected && isThrowUnconnected) {
throw std::runtime_error("Service for Cmd is unavailable yet - try later.");
}
// todo: as monitor-sessions are not distinguished so far, this (direct) exception will not be thrown,
// if a monitor is running and connected
// warn about probably lost samples (if not told otherwise)
if (!isConnected && !isOkUnconnected) {
ddkit::log::info() << "Writer not connected while sending (" << dds::topic::topic_type_name<T>::value() << ").";
}
// do the write
m_writer.write(sample);
}
void dispose(T const& sample) {
DDKIT_LOG_TRACE;
ddkit::log::debug() << "dispose: " << DDKIT_LOG_VAR_VALUE(sample);
// Get handle of instance
auto handle = m_writer.lookup_instance(sample);
// Instance has to be known
if (handle.is_nil()) {
ddkit::log::error() << "Tried to remove object \"" << sample << "\", which does not exist in dds.";
return;
}
// We have to do both `dispose_instance` and `unregister_instance`
// to fully free the instance. If we would just dispose the instance
// the instance would remain stored by the reader and writer since it
// remains `active`. Only after the `unregister_instance` the instance
// has no associated writer and can be freed.
m_writer.dispose_instance(handle);
m_writer.unregister_instance(handle);
}
private:
ConnectStatePtr m_connectState;
std::shared_ptr<ListenerWrite<T>> m_listener;
dds::pub::DataWriter<T> m_writer;
/// Note on writer-listeners:
/// - listeners have to be set in the constructor of the writer, because
/// an event could be missing, when the listener is set afterwards
/// - all depending on how fast discovery etc is, see also here:
/// https://github.com/eclipse-cyclonedds/cyclonedds-cxx/issues/378#issuecomment-1463626212
static dds::pub::DataWriter<T> dataWriter(
std::shared_ptr<Topic<T>> topic,
ListenerWrite<T>* listener,
dds::pub::qos::DataWriterQos dataWriterQos,
dds::pub::qos::PublisherQos pubQos) {
DDKIT_LOG_TRACE;
dds::pub::DataWriter<T> ddsWriter = dds::pub::DataWriter<T>(
dds::pub::Publisher(topic->domainParticipant(), pubQos),
topic->ddsTopic(),
dataWriterQos,
listener,
dds::core::status::StatusMask::all());
return ddsWriter;
}
// no copying
Writer(Writer const&) = delete;
Writer(Writer&&) = delete;
void operator=(Writer const&) = delete;
};
} // namespace access
} // namespace ddkit