qt-migration-legatus/qt6-after/qt6-migrated-legatus/Lts/DDS/DTC/generic_dtc_impl.h

#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() << "Can not set subscriber callback - already set ...";
      return false;
    }
    if (!m_subscriberListener)
    {
      WARNING_LOG() << "Can not set subscriber callback - no subscriber listener ...";
      return false;
    }

    m_subscriberCallback = subscriberCallback;

    if (!m_dataReaderObject)
    {
      WARNING_LOG() << "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() << "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() << "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() << "Error initializing DRO for type " << DDS::topic_type_name<TopicTypeRead>::value();
      delete m_dataReaderObject;
      m_dataReaderObject  = nullptr;
      return false;
    }

    if (m_subscriberListener)
    {
      DEBUG_LOG() << "subscriberListener exists, adding type "
                          << DDS::topic_type_name<TopicTypeRead>::value();
      m_subscriberListener->addCallback(m_dataReaderObject->getDataReader(), this);
    }
    else
    {
      DEBUG_LOG() << "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;
};


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;
};

}