314 lines
10 KiB
C++
Executable File
314 lines
10 KiB
C++
Executable File
|
|
|
|
|
|
#pragma once
|
|
|
|
#include <DDS/Generic/generic_data_reader.h>
|
|
#include <DDS/Generic/subscriber_listener.h>
|
|
#include <QTimer>
|
|
|
|
#include <dds/core/cond/StatusCondition.hpp>
|
|
#include <dds/core/cond/WaitSet.hpp>
|
|
|
|
#include "generic_dtc.h"
|
|
|
|
|
|
namespace DDSGeneric
|
|
{
|
|
class DynamicDataReader;
|
|
}
|
|
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;
|
|
dds::core::cond::StatusCondition cond(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;
|
|
};
|
|
|
|
}
|