#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 class Reader final { public: Reader( std::shared_ptr dispatcher, std::shared_ptr> topic, dds::sub::qos::DataReaderQos dataReaderQos, dds::sub::qos::SubscriberQos subQos, std::function funcWrite, std::function funcDispose) : m_dispatcher(dispatcher) , m_connectState(std::make_shared(topic->topicName(), topic->topicNameModule(), topic->topicNameTopic())) , m_listener(std::make_shared >(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 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 m_dispatcher; ConnectStatePtr m_connectState; std::shared_ptr> m_listener; dds::sub::DataReader m_ddsReader; dds::sub::status::DataState m_dataState; std::atomic m_shutdownRequest; std::function m_funcWrite; std::function m_funcDispose; std::function m_func; dds::sub::cond::ReadCondition m_readCondition; /// Note on reader-listeners: see writer-listeners in Writer.hpp static dds::sub::DataReader dataReader( std::shared_ptr> topic, ListenerRead* listener, dds::sub::qos::DataReaderQos dataReaderQos, dds::sub::qos::SubscriberQos subQos) { DDKIT_LOG_TRACE; dds::sub::DataReader ddsReader = dds::sub::DataReader( 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