qt-migration-legatus/qt5-before/dlbo/Lts/Services/SituationReplay/dds_handler.cpp

/**
  * DdsHandler for DDS
  * $Id$
  * $HeadURL$
  * $Author$
  * Created by Hanna E. Rathje, SLSE, 2016-01-13
  */

#include "dds_handler.h"

#include <DDS/DTS/generic_dts_impl.h>
#include <DDS/QosProfile/DDS_DefaultQos.h>
#include <QVector>

namespace SituationReplay
{

static std::shared_ptr<DDSConnector::DdsConnector> makeConnector(const ConnectionParams &params,
                                                                 IDL::PartitionType partitionType,
                                                                 DDS::QosProfile::ConnectionType defaultConnectionType)
{
  if (params.discoveryPeer.isEmpty())
    return std::make_shared<DDSConnector::DdsConnector>(params.domainId,
                                                        partitionType,
                                                        params.partitionId,
                                                        params.exerciseId,
                                                        defaultConnectionType);
  return std::make_shared<DDSConnector::DdsConnector>(params.domainId,
                                                      partitionType,
                                                      params.partitionId,
                                                      params.exerciseId,
                                                      params.discoveryPeer);
}

DdsHandler::DdsHandler(const ConnectionParams &output)
  : m_ddsConnectorReplay(makeConnector(output, IDL::PartitionType::Replay, DDS::QosProfile::ConnectionType::ShmUdpLocalhost))
  , m_logger(getLogger("DdsHandler"))
{
  DEBUG_LOG(m_logger) << "DdsHandler Replay(" << output.partitionId << ") for partition: " << m_ddsConnectorReplay->getDefaultPartition();
}

DdsHandler::DdsHandler(const ConnectionParams &input, const ConnectionParams &output)
  : m_ddsConnectorExercise(makeConnector(input, IDL::PartitionType::Exercise, DDS::QosProfile::ConnectionType::Client))
  , m_ddsConnectorReplay(makeConnector(output, IDL::PartitionType::Replay, DDS::QosProfile::ConnectionType::Shm))
  , m_logger(getLogger("DdsHandler"))
{
  DEBUG_LOG(m_logger) << "DdsHandler Exercise(" << input.partitionId << ") for partition: " << m_ddsConnectorExercise->getDefaultPartition();
  DEBUG_LOG(m_logger) << "DdsHandler Replay(" << output.partitionId << ") for partition: " << m_ddsConnectorReplay->getDefaultPartition();
}

DdsHandler::~DdsHandler()
{
  m_sessionRouting = nullptr;
  m_dtsPartitionTime = nullptr;
  m_ddsFts = nullptr;
  m_ddsDts = nullptr;
  m_ddsConnectorReplay = nullptr;
  m_ddsConnectorExercise = nullptr;
}

bool DdsHandler::init(const QStringList &qosFileList)
{
  if ((m_ddsConnectorExercise && !m_ddsConnectorExercise->init(qosFileList)) ||
      !m_ddsConnectorReplay->init(qosFileList))
  {
    ERROR_LOG(m_logger) << "Error initializing dds connections - aborting";
    return false;
  }

  if (m_ddsConnectorExercise)
  {
    m_ddsConnectorExercise->initPublisherSubscriber();
    m_ddsConnectorExercise->enableBatchedWriter(false);
  }

  m_ddsConnectorReplay->initPublisherSubscriber();
  m_ddsConnectorReplay->enableBatchedWriter(false);

  m_dtsPartitionTime.reset(m_ddsConnectorReplay->createDTS<DTS::PartitionTimeImpl>());
  assert(m_dtsPartitionTime.get());

  m_ddsDts = SituationReplay::Interface::Dts::create(m_ddsConnectorReplay);
  m_ddsFts = SituationReplay::Interface::Fts::create(m_ddsConnectorReplay);
  bool bRet = m_ddsDts->enableAll();
  INFO_LOG(m_logger) << "Dts initialised";

  bRet &= m_ddsFts->enableAll();
  INFO_LOG(m_logger) << "Fts initialised";

  initReplayRouting();

  return bRet;
}

void DdsHandler::publishPartitionTime(const Core::PartitionTime &partitionTime)
{
  INFO_LOG(m_logger) << "publish partitionTime: " << partitionTime.timeStamp();
  m_dtsPartitionTime->write(partitionTime);
}

bool DdsHandler::parseOptions(const QString &opts, ConnectionParams &params)
{
  QVector<QStringRef> parts = opts.splitRef(QLatin1Char(','));
  if (parts.size() < 2 || parts.size() > 3)
    return false;
  bool bOk;
  params.domainId = parts.at(0).toInt(&bOk);
  if (!bOk)
    return false;
  params.partitionId = parts.at(1).toInt(&bOk);
  if (!bOk)
    return false;
  if (parts.size() == 3)
  {
    QString str = parts.at(2).toString();
    // '_' is not allowed in a scheme, but rti needs it ...
    str.replace(QLatin1String("tcpv4_lan"), QLatin1String("tcpv4lan"));
    QUrl url = QUrl::fromUserInput(str);
    if (url.scheme() == QLatin1String("http") ||
        url.scheme() == QLatin1String("file"))
      url.setScheme(QStringLiteral("tcpv4lan"));
    if (!url.isValid())
      return false;
    params.discoveryPeer = url;
  }
  return true;
}

void DdsHandler::initReplayRouting()
{
  if (!m_ddsConnectorExercise)
    return;
  m_sessionRouting.reset(new RoutingService::SessionHandler(IDL::PartitionType::Replay,
                                                            m_ddsConnectorExercise,
                                                            m_ddsConnectorReplay));
  m_sessionRouting->init();
  m_sessionRouting->onCreateSession(QString::fromStdString(m_ddsConnectorExercise->getDefaultPartition()),
                                    QString::fromStdString(m_ddsConnectorReplay->getDefaultPartition()));
}

} // Namespace SRP