Files
qt6-migration-pipeline/Lts/Common/LtsSession/lts_session_data_dds.cpp
T

390 lines
11 KiB
C++
Executable File

#include "lts_session_data_dds.h"
#include <DDS/DDSConnector/dds_connector.h>
#include <DDS/DTC/generic_dtc_impl.h>
#include <NGLLogger/NGLLoggerBootstrap.h>
#include <QSet>
#include <QVector>
#include <DDS/QosProfile/DDS_DefaultQos.h>
#include <QDebug>
namespace LtsSession
{
class DdsConnectionContainer
{
struct DdsConnection : public DdsConnectionParams
{
std::weak_ptr<DDSConnector::DdsConnector> ddsConnector = std::weak_ptr<DDSConnector::DdsConnector>();
DdsConnection() = default;
explicit DdsConnection(const DdsConnectionParams &p)
: DdsConnectionParams(p)
{}
};
QVector<DdsConnection> ddsConnections;
public:
std::shared_ptr<DDSConnector::DdsConnector> findGlobalConnection()
{
for (int i = ddsConnections.size() - 1; i >= 0; --i)
{
const DdsConnection &conn = ddsConnections.at(i);
if (conn.partitionType != IDL::PartitionType::Global)
continue;
auto ret = conn.ddsConnector.lock();
if (ret != nullptr)
return ret;
ddsConnections.removeAt(i);
}
Q_ASSERT(false);
return std::shared_ptr<DDSConnector::DdsConnector>();
}
std::shared_ptr<DDSConnector::DdsConnector> findConnection(const DdsConnectionParams &connParams)
{
for (int i = ddsConnections.size() - 1; i >= 0; --i)
{
const DdsConnection &conn = ddsConnections.at(i);
if (conn != connParams)
continue;
auto ret = conn.ddsConnector.lock();
if (ret != nullptr)
return ret;
ddsConnections.removeAt(i);
}
return std::shared_ptr<DDSConnector::DdsConnector>();
}
dds::domain::DomainParticipant findMatchingDomainParticipant(const DdsConnectionParams &connParams)
{
for (int i = ddsConnections.size() - 1; i >= 0; --i)
{
const DdsConnection &conn = ddsConnections.at(i);
if (!connParams.hasSameDomainParams(conn))
continue;
auto ret = conn.ddsConnector.lock();
if (ret != nullptr)
return ret->getDomainParticipant();
ddsConnections.removeAt(i);
}
return nullptr;
}
void insertConnection(const DdsConnectionParams &connParams,
const std::shared_ptr<DDSConnector::DdsConnector> &ddsConnector)
{
DdsConnection val(connParams);
val.ddsConnector = ddsConnector;
ddsConnections += val;
}
};
Q_GLOBAL_STATIC(DdsConnectionContainer, s_ddsConnections)
class DataPluginContainer
{
public:
struct DataPluginValue
{
QString identifier;
std::weak_ptr<DataPluginBase::Interface> weakIfc;
bool operator==(const DataPluginValue &o) const
{
return identifier == o.identifier && weakIfc.lock().get() == o.weakIfc.lock().get();
}
};
QMap<DdsConnectionParams, QSet<DataPluginValue>> m_dataPlugins;
public:
QMap<QString, std::shared_ptr<DataPluginBase::Interface>> findDataPlugins(const DdsConnectionParams &connParams);
void insertDataPlugins(const DdsConnectionParams &connParams,
const QMap<QString, std::shared_ptr<DataPluginBase::Interface>> &dataPlugins);
};
uint qHash(const DataPluginContainer::DataPluginValue &val)
{
qint64 ptr = reinterpret_cast<qint64>(val.weakIfc.lock().get());
return ::qHash(val.identifier) ^ ::qHash(ptr);
}
QMap<QString, std::shared_ptr<DataPluginBase::Interface>> DataPluginContainer::findDataPlugins(const DdsConnectionParams &connParams)
{
QMap<QString, std::shared_ptr<DataPluginBase::Interface>> ret;
DdsConnectionParams key(connParams);
key.exerciseId = 0;
auto it = m_dataPlugins.find(key);
if (it != m_dataPlugins.end())
{
QSet<DataPluginValue> &cont = it.value();
auto itCont = cont.begin();
while (itCont != cont.end())
{
auto spPlugin = itCont->weakIfc.lock();
if (spPlugin)
{
ret[itCont->identifier] = spPlugin;
++itCont;
}
else
{
itCont = cont.erase(itCont);
}
}
}
return ret;
}
void DataPluginContainer::insertDataPlugins(const DdsConnectionParams &connParams,
const QMap<QString, std::shared_ptr<DataPluginBase::Interface>> &dataPlugins)
{
DdsConnectionParams key(connParams);
key.exerciseId = 0;
QSet<DataPluginValue> cont;
cont.reserve(dataPlugins.size());
for (auto it = dataPlugins.begin(); it != dataPlugins.end(); ++it)
{
DataPluginValue val;
val.identifier = it.key();
val.weakIfc = it.value();
cont += val;
}
m_dataPlugins[key].unite(cont);
}
Q_GLOBAL_STATIC(DataPluginContainer, s_dataPlugins)
LtsSessionDataDds::LtsSessionDataDds(QObject *parent)
: QObject(parent)
, m_exerciseId(-1)
, m_emittedTime(0)
, m_curMSecSinceEpoch(0)
, m_active(true)
, m_fetcher(nullptr)
, m_logger(NGL::NGLLoggerBootstrap::getNGLRootLogger().getNGLLogger("LtsSession.Session"))
{}
LtsSessionDataDds::~LtsSessionDataDds()
{
m_dtcPartitionTime = nullptr;
m_ddsConnector = nullptr;
delete m_fetcher;
}
bool LtsSessionDataDds::init(const DdsConnectionParams &connParams)
{
m_ddsConnector = s_ddsConnections->findConnection(connParams);
m_exerciseId = connParams.exerciseId;
m_qosFileList = QStringList(connParams.qos_file);
if (m_ddsConnector == nullptr)
{
dds::domain::DomainParticipant dp = s_ddsConnections->findMatchingDomainParticipant(connParams);
if (dp != nullptr)
return init(dp, connParams);
#ifdef WIN_32
DDS::QosProfile::ConnectionType type = DDS::QosProfile::ConnectionType::Tablet;
#else
DDS::QosProfile::ConnectionType type = DDS::QosProfile::ConnectionType::Client;
#endif
m_ddsConnector = std::make_shared<DDSConnector::DdsConnector>(connParams.domainId,
connParams.partitionType,
connParams.partitionId,
connParams.exerciseId,
type);
s_ddsConnections()->insertConnection(connParams, m_ddsConnector);
}
return initInternal();
}
bool LtsSessionDataDds::init(const dds::domain::DomainParticipant &dp, const DdsConnectionParams &connParams)
{
Q_ASSERT(dp.domain_id() == connParams.domainId);
m_ddsConnector = s_ddsConnections->findConnection(connParams);
m_exerciseId = connParams.exerciseId;
QStringList qosList;
if (!connParams.qos_file.isEmpty())
{
qosList.append(connParams.qos_file);
}
if (!qosList.isEmpty())
m_qosFileList = qosList;
else
m_qosFileList.clear();
#ifdef WIN_32
DDS::QosProfile::ConnectionType type = DDS::QosProfile::ConnectionType::Tablet;
#else
DDS::QosProfile::ConnectionType type = DDS::QosProfile::ConnectionType::Client;
#endif
if (m_ddsConnector == nullptr)
{
m_ddsConnector = std::make_shared<DDSConnector::DdsConnector>(dp,
connParams.partitionType,
type,
connParams.partitionId
);
s_ddsConnections()->insertConnection(connParams, m_ddsConnector);
}
return initInternal();
}
Fetcher *LtsSessionDataDds::fetcher(const QString &yourPluginsName)
{
if (m_fetcher == nullptr)
m_fetcher = new Fetcher(this, yourPluginsName);
return m_fetcher;
}
QString LtsSessionDataDds::partitionTypeString(IDL::PartitionType partitionType)
{
switch (partitionType)
{
case IDL::PartitionType::AllExercises:
return QStringLiteral("AllExercises");
case IDL::PartitionType::Any:
return QStringLiteral("Any");
case IDL::PartitionType::Exercise:
return QStringLiteral("Exercise");
case IDL::PartitionType::Global:
return QStringLiteral("Global");
case IDL::PartitionType::Replay:
return QStringLiteral("Replay");
}
return QStringLiteral("Invalid (%1)").arg(int(partitionType));
}
QString LtsSessionDataDds::toString() const
{
return QStringLiteral("(LtsSessionDataDds: id=%1, partitionId=%2, partitionType=%3(%4), exerciseId=%5)").arg((qulonglong)this).arg(partitionId()).arg(static_cast<int>(partitionType())).arg(partitionTypeString(partitionType())).arg(exerciseId());
}
int LtsSessionDataDds::domainId() const
{
return m_ddsConnector ? m_ddsConnector->getDomainId() : -1;
}
int LtsSessionDataDds::exerciseId() const
{
return m_exerciseId;
}
int LtsSessionDataDds::partitionId() const
{
return m_ddsConnector ? m_ddsConnector->getPartitionId() : -1;
}
IDL::PartitionType LtsSessionDataDds::partitionType() const
{
return m_ddsConnector ? m_ddsConnector->getPartitionType() : IDL::PartitionType::Global;
}
QStringList LtsSessionDataDds::qosFile() const
{
return m_ddsConnector ? m_qosFileList : QStringList();
}
QUrl LtsSessionDataDds::tcpConnectionUrl() const
{
return m_ddsConnector ? m_ddsConnector->getTcpConnectionUrl() : QUrl();
}
DdsConnectionParams LtsSessionDataDds::connParams() const
{
DdsConnectionParams connParams;
connParams.domainId = domainId();
connParams.exerciseId = exerciseId();
connParams.partitionId = partitionId();
connParams.partitionType = partitionType();
connParams.qos_file = qosFile();
connParams.tcpConnectionUrl = tcpConnectionUrl();
return connParams;
}
bool LtsSessionDataDds::initInternal()
{
if (!m_ddsConnector || !m_ddsConnector->init(m_qosFileList))
{
ERROR_LOG(m_logger) << "DDS connector could not be initialized - session aborted!";
m_ddsConnector = nullptr;
return false;
}
m_ddsConnector->initPublisherSubscriber();
m_ddsConnectorGlobal = s_ddsConnections->findGlobalConnection();
if (!m_ddsConnectorGlobal)
m_ddsConnectorGlobal = m_ddsConnector;
m_dtcPartitionTime.reset(m_ddsConnector->createDTC<DTC::PartitionTimeImpl>());
assert(m_dtcPartitionTime.get());
m_dtcPartitionTime->setSubscriberCallback([this](const Core::PartitionTime & data, DataModification)
{
m_curMSecSinceEpoch = data.timeStamp();
emit partitionTimeChangedInternal();
});
connect(this, &LtsSessionDataDds::partitionTimeChangedInternal,
this, &LtsSessionDataDds::onPartitionTimeChangedInternal, Qt::QueuedConnection);
return true;
}
void LtsSessionDataDds::onPartitionTimeChangedInternal()
{
if (!m_active)
return;
if (m_emittedTime == m_curMSecSinceEpoch)
return;
m_emittedTime = m_curMSecSinceEpoch;
emit partitionTimeChanged(m_curMSecSinceEpoch);
}
QMap<QString, std::shared_ptr<DataPluginBase::Interface>> LtsSessionDataDds::availableDataPluginInterfaces() const
{
return s_dataPlugins->findDataPlugins(connParams());
}
void LtsSessionDataDds::addAvailablaeDataPluginInterfaces(const QMap<QString, std::shared_ptr<DataPluginBase::Interface>> &dataPlugins)
{
s_dataPlugins->insertDataPlugins(connParams(),
dataPlugins);
}
}