301 lines
11 KiB
C++
301 lines
11 KiB
C++
|
|
|
|
#pragma once
|
|
|
|
// Include Qt headers FIRST to prevent namespace pollution when RTI DDS headers include Qt
|
|
#include <QtCore/QString>
|
|
#include <QtCore/QUrl>
|
|
#include <QtCore/QFlag>
|
|
#include <QtCore/QTimer>
|
|
#include "generic_dtc.h"
|
|
#include <DDS/Generic/generic_data_reader.h>
|
|
#include <DDS/Generic/generic_logger.h>
|
|
#include <DDS/Generic/subscriber_listener.h>
|
|
#include <dds/core/cond/StatusCondition.hpp>
|
|
#include <dds/core/cond/WaitSet.hpp>
|
|
|
|
namespace DDSGeneric
|
|
{
|
|
class DynamicDataReader;
|
|
} // namespace DDSGeneric
|
|
|
|
namespace DTC
|
|
{
|
|
|
|
template <typename SuperClass>
|
|
class GenericDtcImplSubscriberTemplate : public DDSGeneric::SubscriberListenerCallback, public SuperClass
|
|
{
|
|
public:
|
|
using DataReaderObject = typename SuperClass::DataReaderObject;
|
|
using SubscriberCallback = typename SuperClass::SubscriberCallback;
|
|
using TopicTypeRead = typename DataReaderObject::TopicType;
|
|
using TopicTypeSubscribe = typename DataReaderObject::TopicType;
|
|
using DataReader = typename DataReaderObject::DataReader;
|
|
|
|
GenericDtcImplSubscriberTemplate(
|
|
const dds::domain::DomainParticipant& domainParticipant, const dds::sub::Subscriber& ddsSubscriber,
|
|
DDSGeneric::SubscriberListener* subscriberListener,
|
|
const std::string& topicName = DDS::Helper::getShortTopicName(DDS::topic_type_name<TopicTypeRead>::value()))
|
|
: m_subscriberListener(subscriberListener), m_dataReaderObject(new DataReaderObject),
|
|
m_subscriberCallback(nullptr), m_writerState(0),
|
|
m_logger(DDS::getLogger(DDS::topic_type_name<TopicTypeRead>::value(), DDS::DataType::DATA_TOPIC_CLIENT))
|
|
{
|
|
createInternal(m_dataReaderObject->create(domainParticipant, ddsSubscriber, topicName));
|
|
}
|
|
|
|
GenericDtcImplSubscriberTemplate(const dds::domain::DomainParticipant& domainParticipant,
|
|
const dds::sub::Subscriber& ddsSubscriber,
|
|
DDSGeneric::SubscriberListener* subscriberListener,
|
|
const dds::topic::Filter& filter, const std::string& topicName)
|
|
: m_subscriberListener(subscriberListener), m_dataReaderObject(new DataReaderObject),
|
|
m_subscriberCallback(nullptr), m_writerState(0),
|
|
m_logger(DDS::getLogger(DDS::topic_type_name<TopicTypeRead>::value(), DDS::DataType::DATA_TOPIC_CLIENT))
|
|
{
|
|
createInternal(m_dataReaderObject->create(domainParticipant, ddsSubscriber, topicName, filter));
|
|
}
|
|
|
|
GenericDtcImplSubscriberTemplate(const dds::domain::DomainParticipant& domainParticipant,
|
|
const dds::sub::Subscriber& ddsSubscriber,
|
|
DDSGeneric::SubscriberListener* subscriberListener,
|
|
const dds::core::xtypes::StructType& structType, const std::string& topicName)
|
|
: m_subscriberListener(subscriberListener), m_dataReaderObject(new DataReaderObject),
|
|
m_subscriberCallback(nullptr),
|
|
m_logger(DDS::getLogger(DDS::topic_type_name<TopicTypeRead>::value(), DDS::DataType::DATA_TOPIC_CLIENT))
|
|
{
|
|
createInternal(m_dataReaderObject->create(domainParticipant, ddsSubscriber, topicName, structType));
|
|
}
|
|
|
|
~GenericDtcImplSubscriberTemplate()
|
|
{
|
|
if (m_dataReaderObject && m_subscriberCallback)
|
|
m_subscriberListener->removeCallback(m_dataReaderObject->getDataReader(), this);
|
|
delete m_dataReaderObject;
|
|
}
|
|
|
|
bool setSubscriberCallback(const SubscriberCallback& subscriberCallback) override final
|
|
{
|
|
if (m_subscriberCallback)
|
|
{
|
|
WARNING_LOG(m_logger) << "Can not set subscriber callback - already set ...";
|
|
return false;
|
|
}
|
|
if (!m_subscriberListener)
|
|
{
|
|
WARNING_LOG(m_logger) << "Can not set subscriber callback - no subscriber listener ...";
|
|
return false;
|
|
}
|
|
|
|
m_subscriberCallback = subscriberCallback;
|
|
|
|
if (!m_dataReaderObject)
|
|
{
|
|
WARNING_LOG(m_logger) << "Can not set subscriber callback - no data reader object ...";
|
|
return false;
|
|
}
|
|
|
|
m_dataReaderObject->setHaveSubscriberCallback(m_subscriberCallback != nullptr);
|
|
|
|
return true;
|
|
}
|
|
|
|
dds::sub::LoanedSamples<TopicTypeRead> readAlive() const override final
|
|
{
|
|
if (!m_dataReaderObject)
|
|
return dds::sub::LoanedSamples<TopicTypeRead>();
|
|
|
|
bool bSeenUnreadSamples = false;
|
|
auto ret = m_dataReaderObject->readAlive(bSeenUnreadSamples);
|
|
if (bSeenUnreadSamples)
|
|
QTimer::singleShot(0, this, &GenericDtcImplSubscriberTemplate::unreadSamplesDuringRead);
|
|
|
|
return ret;
|
|
}
|
|
|
|
dds::sub::LoanedSamples<TopicTypeRead> takeAlive() const override final
|
|
{
|
|
if (!m_dataReaderObject)
|
|
return dds::sub::LoanedSamples<TopicTypeRead>();
|
|
|
|
bool bSeenUnreadSamples = false;
|
|
auto ret = m_dataReaderObject->takeAlive(bSeenUnreadSamples);
|
|
if (bSeenUnreadSamples)
|
|
QTimer::singleShot(0, this, &GenericDtcImplSubscriberTemplate::unreadSamplesDuringRead);
|
|
|
|
return ret;
|
|
}
|
|
|
|
dds::sub::LoanedSamples<TopicTypeRead> takeDisposed() const override final
|
|
{
|
|
if (!m_dataReaderObject)
|
|
return dds::sub::LoanedSamples<TopicTypeRead>();
|
|
|
|
return m_dataReaderObject->takeDisposed();
|
|
}
|
|
|
|
bool hasSamples() const override final
|
|
{
|
|
if (!m_dataReaderObject)
|
|
return false;
|
|
|
|
return m_dataReaderObject->hasSamples();
|
|
}
|
|
|
|
DataReaderObject* dataReader() const override final
|
|
{
|
|
return m_dataReaderObject;
|
|
}
|
|
|
|
bool waitForHistoricalData(int timeoutMSec) override
|
|
{
|
|
if (m_dataReaderObject == nullptr || m_dataReaderObject->getDataReader() == nullptr)
|
|
return false;
|
|
|
|
const bool bWaitForever = (timeoutMSec < 0);
|
|
auto dr = m_dataReaderObject->getDataReader();
|
|
try
|
|
{
|
|
if (dr.subscription_matched_status().current_count() == 0)
|
|
{
|
|
dds::core::cond::WaitSet ws;
|
|
auto cond = dds::core::cond::StatusCondition(dr);
|
|
cond.enabled_statuses(dds::core::status::StatusMask::subscription_matched());
|
|
ws.attach_condition(cond);
|
|
const dds::core::Duration waitTimeout(bWaitForever ? dds::core::Duration::infinite()
|
|
: dds::core::Duration::from_millisecs(timeoutMSec));
|
|
const dds::core::cond::WaitSet::ConditionSeq seq(ws.wait(waitTimeout));
|
|
|
|
if (seq.empty())
|
|
return false;
|
|
}
|
|
dr.wait_for_historical_data(dds::core::Duration(1));
|
|
}
|
|
catch (const dds::core::Exception&)
|
|
{
|
|
|
|
return false;
|
|
}
|
|
return true;
|
|
}
|
|
|
|
protected:
|
|
void unreadSamplesDuringRead()
|
|
{
|
|
droDataAvailable(DDS::topic_type_name<TopicTypeRead>::value(), m_dataReaderObject->topicName());
|
|
}
|
|
|
|
void droDataAvailable(const std::string& topicType, const std::string& topicName) override final
|
|
{
|
|
Q_UNUSED(topicType);
|
|
if (m_writerState.testAndSetRelaxed(1, 0))
|
|
{
|
|
DEBUG_LOG(m_logger) << "Recreate Reader for topic " << topicName << ", type name "
|
|
<< DDS::topic_type_name<TopicTypeRead>::value();
|
|
if (m_subscriberListener)
|
|
m_subscriberListener->removeCallback(m_dataReaderObject->getDataReader(), this);
|
|
m_dataReaderObject->recreateReader();
|
|
if (m_subscriberListener)
|
|
m_subscriberListener->addCallback(m_dataReaderObject->getDataReader(), this);
|
|
}
|
|
|
|
DEBUG_LOG(m_logger) << "IN droDataAvailable(), topic name: " << topicName << ", type name "
|
|
<< DDS::topic_type_name<TopicTypeRead>::value() << ", this: " << this
|
|
<< ", m_dataReaderObject: " << m_dataReaderObject
|
|
<< ", hasSamples: " << (m_dataReaderObject ? m_dataReaderObject->hasSamples() : -1);
|
|
|
|
if (m_subscriberListener)
|
|
m_subscriberListener->enableCallback(m_dataReaderObject->getDataReader(), this);
|
|
|
|
if (!m_dataReaderObject || !m_subscriberCallback)
|
|
return;
|
|
|
|
m_dataReaderObject->read(m_subscriberCallback);
|
|
}
|
|
|
|
void noWriterAnymore(const std::string& topicType, const std::string& topicName) override
|
|
{
|
|
Q_UNUSED(topicType);
|
|
Q_UNUSED(topicName);
|
|
m_writerState = -1;
|
|
}
|
|
|
|
void writerAvailable(const std::string& topicType, const std::string& topicName) override
|
|
{
|
|
Q_UNUSED(topicType);
|
|
Q_UNUSED(topicName);
|
|
m_writerState.testAndSetRelaxed(-1, 1);
|
|
}
|
|
|
|
bool createInternal(bool droInitOk)
|
|
{
|
|
if (!droInitOk)
|
|
{
|
|
ERROR_LOG(m_logger) << "Error initializing DRO for type " << DDS::topic_type_name<TopicTypeRead>::value();
|
|
delete m_dataReaderObject;
|
|
m_dataReaderObject = nullptr;
|
|
return false;
|
|
}
|
|
|
|
if (m_subscriberListener)
|
|
{
|
|
DEBUG_LOG(m_logger) << "subscriberListener exists, adding type "
|
|
<< DDS::topic_type_name<TopicTypeRead>::value();
|
|
m_subscriberListener->addCallback(m_dataReaderObject->getDataReader(), this);
|
|
}
|
|
else
|
|
{
|
|
DEBUG_LOG(m_logger) << "Cannot add DTC to subscriberListener callback - type: "
|
|
<< DDS::topic_type_name<TopicTypeRead>::value();
|
|
}
|
|
m_dataReaderObject->setHaveSubscriberCallback(m_subscriberListener != nullptr);
|
|
return true;
|
|
}
|
|
|
|
private:
|
|
DDSGeneric::SubscriberListener* m_subscriberListener;
|
|
|
|
DataReaderObject* m_dataReaderObject;
|
|
|
|
SubscriberCallback m_subscriberCallback;
|
|
|
|
QAtomicInt m_writerState;
|
|
|
|
NGL::INGLLogger& m_logger;
|
|
};
|
|
|
|
template <typename DRO_OBJECT>
|
|
class GenericDtcImpl final : public GenericDtcImplSubscriberTemplate<GenericDtc<DRO_OBJECT>>
|
|
{
|
|
using BaseClass = GenericDtcImplSubscriberTemplate<GenericDtc<DRO_OBJECT>>;
|
|
|
|
public:
|
|
using BaseClass::BaseClass;
|
|
};
|
|
|
|
template <typename DRO_OBJECT>
|
|
class GenericDtcImplSampleInfo final : public GenericDtcImplSubscriberTemplate<GenericDtcSampleInfo<DRO_OBJECT>>
|
|
{
|
|
using BaseClass = GenericDtcImplSubscriberTemplate<GenericDtcSampleInfo<DRO_OBJECT>>;
|
|
|
|
public:
|
|
using BaseClass::BaseClass;
|
|
};
|
|
|
|
template <typename DRO_OBJECT> class DdtcImpl final : public GenericDtcImplSubscriberTemplate<Ddtc<DRO_OBJECT>>
|
|
{
|
|
using BaseClass = GenericDtcImplSubscriberTemplate<Ddtc<DRO_OBJECT>>;
|
|
|
|
public:
|
|
using BaseClass::BaseClass;
|
|
};
|
|
|
|
template <typename DRO_OBJECT>
|
|
class DdtcImplSampleInfo final : public GenericDtcImplSubscriberTemplate<DdtcSampleInfo<DRO_OBJECT>>
|
|
{
|
|
using BaseClass = GenericDtcImplSubscriberTemplate<DdtcSampleInfo<DRO_OBJECT>>;
|
|
|
|
public:
|
|
using BaseClass::BaseClass;
|
|
};
|
|
|
|
} // namespace DTC
|