#include "PrtCnvCentralThread.hpp" #include "PrtCnvRadioThread.hpp" #include "PrtCnvDbThread.hpp" #include "PrtCnvStoredCentralTelegram.hpp" #include "PrtCnvTelegramLogThread.hpp" #include "PrtCnvInjectionThread.hpp" #include "PrtCnvThreadStarter.hpp" #include #include #include #include #include #include #include #include #include #include namespace PrtCnv { PrtCnvCentralThread::PrtCnvCentralThread(PrtCnv::PrtCnvThreadStarter* a_parent) : PrtCnvThread(a_parent, "PrtCnvCentralThread") , m_webSocket(nullptr) , m_SendCounter(0) , m_RadioThread(nullptr) , m_RadioStopped(false) , m_CentralStopped(false) , m_LastSendCounter(0) , m_VirtualTeilnehmerId(0) , m_maxEqID(50000) , m_aliveTimer(nullptr) , m_sendAckTimer(nullptr) , m_reconnectTimer(nullptr) , m_DbThread(nullptr) , m_LogThread(nullptr) , m_InjectionThread(nullptr) , m_ReceiveCounter(0) , m_ReceiveAckCounter(0) , m_State(PrtCnvSocketThread::e_INITIAL) , m_bDirectConnection(false) , m_bRadioThreadInitDone(false) { m_Name = QStringLiteral("Central"); } PrtCnvCentralThread::~PrtCnvCentralThread() { destroy(); } int32_t PrtCnvCentralThread::create(NGL::Settings *settings) { PrtCnvThread::create(settings); INFO_LOG(m_logger) << "Create"; m_RadioThread = qobject_cast(m_Parent->getOtherThread(PrtCnvApplication::e_Radio)); m_DbThread = qobject_cast(m_Parent->getOtherThread(PrtCnvApplication::e_Db)); m_LogThread = qobject_cast(m_Parent->getOtherThread(PrtCnvApplication::e_Log)); m_InjectionThread = qobject_cast(m_Parent->getOtherThread(PrtCnvApplication::e_Injection)); connect(m_DbThread, &PrtCnvDbThread::playerLicencedChanged, this, &PrtCnvCentralThread::onPlayerLicencedChanged, Qt::QueuedConnection); connect(this, &PrtCnvCentralThread::checkSendToCentral, this, &PrtCnvCentralThread::checkSend, Qt::QueuedConnection); connect(this, &PrtCnvCentralThread::checkSendToRadio, m_RadioThread, &PrtCnvRadioThread::checkSend, Qt::QueuedConnection); connect(this, &PrtCnvCentralThread::stopRadio, m_RadioThread, &PrtCnvRadioThread::doStop, Qt::QueuedConnection); connect(this, &PrtCnvCentralThread::cleanRadio, m_RadioThread, &PrtCnvRadioThread::doClean, Qt::QueuedConnection); connect(this, &PrtCnvCentralThread::startRadio, m_RadioThread, &PrtCnvRadioThread::doStart, Qt::QueuedConnection); connect(m_RadioThread, &PrtCnvRadioThread::initDone, this, &PrtCnvCentralThread::onRadioThreadInitDone); connect(m_RadioThread, &PrtCnvRadioThread::areaEffectSimulationActivated, this, &PrtCnvCentralThread::onAreaEffectSimulationActivated, Qt::QueuedConnection); connect(m_RadioThread, &PrtCnvRadioThread::areaEffectSimulationDeactivated, this, &PrtCnvCentralThread::onAreaEffectSimulationDeactivated, Qt::QueuedConnection); auto mainApp = m_Parent->getMainApp(); connect(this, &PrtCnvCentralThread::directTextMessage, mainApp, &PrtCnvApplication::textMessageFromPrtCnvToCentral); connect(mainApp, &PrtCnvApplication::textMessageFromCentralToPrtCnv, this, &PrtCnvCentralThread::onDirectTextMessageReceived); m_aliveTimer = new QTimer(this); connect(m_aliveTimer, &QTimer::timeout, this, &PrtCnvCentralThread::timeoutAliveAndConnection); m_sendAckTimer = new QTimer(this); connect(m_sendAckTimer, &QTimer::timeout, this, &PrtCnvCentralThread::timeoutTick); const int disconnectTimeout = m_Config->getInteger(getName(), QStringLiteral("DisconnectTimeout"), 10000); m_reconnectTimer = new QTimer(this); m_reconnectTimer->setInterval(disconnectTimeout); m_reconnectTimer->setSingleShot(true); connect(m_reconnectTimer, &QTimer::timeout, this, &PrtCnvCentralThread::connectSocket); return 0; } int32_t PrtCnvCentralThread::destroy() { INFO_LOG(m_logger) << "Destroy"; PrtCnvThread::destroy(); delete m_webSocket; m_webSocket = nullptr; delete m_aliveTimer; m_aliveTimer = nullptr; delete m_sendAckTimer; m_sendAckTimer = nullptr; delete m_reconnectTimer; m_reconnectTimer = nullptr; return 0; } int32_t PrtCnvCentralThread::init() { INFO_LOG(m_logger) << "Init"; PrtCnvThread::init(); m_CentralStopped = false; m_RadioStopped = false; m_aliveTimer->setInterval(m_Config->getInteger(getName(), QStringLiteral("AliveTimeout"), 3000)); m_sendAckTimer->setInterval(m_Config->getInteger(getName(), QStringLiteral("AckTimeout"), 900)); m_VirtualTeilnehmerId = m_Config->getInteger(QStringLiteral("Common"), QStringLiteral("VirtualIdPrtCnv"), 90); m_maxEqID = m_Config->getInteger(QStringLiteral("Common"), QStringLiteral("MaxEqID"), 50000); return 0; } int32_t PrtCnvCentralThread::deinit() { INFO_LOG(m_logger) << "Deinit"; PrtCnvThread::deinit(); return 0; } int32_t PrtCnvCentralThread::start() { INFO_LOG(m_logger) << "Start"; PrtCnvThread::start(); if (m_DbThread->mustProceedCleanup()) { INFO_LOG(m_logger) << "Proceed cleanup DB after crash..."; m_CentralStopped = true; emit stopRadio(); } else { connectSocket(); } return 0; } int32_t PrtCnvCentralThread::stop() { INFO_LOG(m_logger) << "Stop"; PrtCnvThread::stop(); return 0; } void PrtCnvCentralThread::connected() { INFO_LOG(m_logger) << "Central-Socket connected m_Socket=" << m_webSocket; m_SendCounter = 0; m_LastSendCounter = 0; m_ReceiveCounter = 0; m_ReceiveAckCounter = 0; m_aliveTimer->start(); if (m_bRadioThreadInitDone) m_sendAckTimer->start(); m_State = PrtCnvSocketThread::e_WORKING; emit checkSendToCentral(); } void PrtCnvCentralThread::disconnected() { INFO_LOG(m_logger) << "disconnected m_Socket=" << m_webSocket; if (PrtCnvSocketThread::e_WORKING == m_State && m_webSocket) m_webSocket->close(); m_State = PrtCnvSocketThread::e_DISCONNECTING; if (m_DoCleanup) { INFO_LOG(m_logger) << "cleanup while disconnection"; m_sendAckTimer->stop(); m_CentralStopped = true; if (m_RadioStopped) cleanAndReinit(); } else { m_DbThread->prepareCentralTelegramsForReconnection(); connectSocket(); } } void PrtCnvCentralThread::radioStopped() { m_RadioStopped = true; if (m_CentralStopped) cleanAndReinit(); } void PrtCnvCentralThread::cleanAndReinit() { INFO_LOG(m_logger) << "cleanAndReinit"; emit cleanRadio(); m_ReceiveAckCounter = 0; m_DbThread->cleanAndInit(); m_DoCleanup = false; } void PrtCnvCentralThread::radioCleant() { if (!m_DoCleanup) { m_DbThread->setMachineState(0); m_RadioStopped = false; m_CentralStopped = false; emit startRadio(); connectSocket(); } else { ERROR_LOG(m_logger) << "Internal error: radioCleant, but central still cleaning up."; } } void PrtCnvCentralThread::onTextMessageReceived(const QString &msg) { if (m_DoCleanup) return; if (PrtCnvSocketThread::e_WORKING != m_State) return; onDirectTextMessageReceived(msg.toUtf8()); } void PrtCnvCentralThread::onDirectTextMessageReceived(const QByteArray &baJson) { try { if (baJson.isEmpty()) throw std::exception(); const QJsonObject obj = QJsonDocument::fromJson(baJson).object(); const QJsonValue typeId = obj.value(QStringLiteral("typeId")); if (!typeId.isDouble()) throw TrainingAreaBase::TetraIcd::JsonException(obj, QStringLiteral("typeId")); N_CoPrtCnv::CoPrtCnvCentralTelegram telegram(baJson, obj, static_cast(typeId.toInt())); m_LogThread->writeCentralTelegramLog(&telegram, false, 1); m_aliveTimer->start(); if (telegram.telegramId() == TrainingAreaBase::TetraIcd::BESTAETIGUNG) { handleAck(&telegram); } else { DEBUG_LOG(m_logger) << "Received telegram '" << baJson << "'"; handleTelegram(&telegram); } } catch (const std::exception &e) { ERROR_LOG(m_logger) << "JSon conversion error happenend. Error:" << e.what() << ", msg: " << baJson; TrainingAreaBase::TetraIcd::CtaDataError tele; tele.header.equipmentNumber = m_VirtualTeilnehmerId; tele.header.timeStamp = time(nullptr) * 1000LL; tele.reason = TrainingAreaBase::TetraIcd::CtaDataError::Reason::JsonDecodeFailure; tele.originTelegramId = 0; N_CoPrtCnv::CoPrtCnvCentralTelegram telegram(tele); sendTelegram(&telegram); } } void PrtCnvCentralThread::stateChanged(QAbstractSocket::SocketState newState) { INFO_LOG(m_logger) << "Websocket changed to state " << newState; } void PrtCnvCentralThread::handleTelegram(N_CoPrtCnv::CoPrtCnvCentralTelegram *a_telegram) { DEBUG_LOG(m_logger) << "Receive Central-Telegram with telId=" << a_telegram->telegramId() << "; rcounter=" << a_telegram->counter(); if (checkAndUpdateReceiveCounter(a_telegram->counter())) { if (handleSpecialTelegram(a_telegram)) { using namespace TrainingAreaBase::TetraIcd; if (a_telegram->telegramId() == TelegramId::DEAKTIVIERUNG_WIRKFELD) { const DeactivateEffectSimulation des(DeactivateEffectSimulation::fromJson(a_telegram->jsonData())); switch (des.aesType) { case AreaEffectSimulationType::MineArea: case AreaEffectSimulationType::ExplosionArea: case AreaEffectSimulationType::ExplosionSingle: case AreaEffectSimulationType::NContamination: case AreaEffectSimulationType::CContamination: case AreaEffectSimulationType::FunctionalArea: case AreaEffectSimulationType::DusSingleMineSimulator: if (m_deactivatedAreaEffectSimulations.contains(des.aesDdsId)) { INFO_LOG(m_logger) << "AES with id " << des.aesDdsId << " already deactivated - ignoring new deactivation telegram"; return; } break; case AreaEffectSimulationType::TechnicalArea: case AreaEffectSimulationType::GasPipeDetonation: case AreaEffectSimulationType::SpecialArea: case AreaEffectSimulationType::Unknown: Q_ASSERT(false); break; case AreaEffectSimulationType::BeStillWarning: break; } } else if (a_telegram->telegramId() == TelegramId::WIRKFELD_STATUS_MINENSPERRE) { const MineAreaSingleMineState msms(MineAreaSingleMineState::fromJson(a_telegram->jsonData())); QSet &oldTriggeredMines = m_alreadyTriggeredMines[msms.aesDdsId]; QSet newTriggeredMines; newTriggeredMines.reserve(oldTriggeredMines.size() + 5); bool bSomethingChanged = false; for (int i = 0; i < msms.mineCount; ++i) { if (!LtsUtils::BitArray::testBit(msms.mineState, i)) { INFO_LOG(m_logger) << "Testing bit " << i << " ... not set = triggered"; if (!oldTriggeredMines.contains(i)) { INFO_LOG(m_logger) << "not in old one"; bSomethingChanged = true; } newTriggeredMines.insert(i); } } if (!bSomethingChanged) bSomethingChanged = newTriggeredMines.size() != oldTriggeredMines.size(); if (!bSomethingChanged) { INFO_LOG(m_logger) << "MineState telegram contains no new informations - skipping " << msms.aesDdsId; return; } INFO_LOG(m_logger) << "MineState telegram contains new informations... " << msms.aesDdsId; oldTriggeredMines = newTriggeredMines; } std::shared_ptr radioTelegram(N_CoPrtCnv::CoPrtCnvTelegramConverter::convertToRadio(a_telegram)); if (radioTelegram) { if (radioTelegram->getIdentity() <= m_maxEqID) { m_DbThread->storeRadioTelegram(radioTelegram, a_telegram->counter()); } else { WARNING_LOG(m_logger) << "Central telegram with telId=" << a_telegram->telegramId() << " has invalid identity:" << radioTelegram->getIdentity(); } } else { m_DbThread->centralCounterReadyForAck(a_telegram->counter()); WARNING_LOG(m_logger) << "Central telegram with telId=" << a_telegram->telegramId() << " could not be converted"; } } } emit checkSendToRadio(); } bool PrtCnvCentralThread::checkAndUpdateReceiveCounter(const int32_t a_counter) { TRACE_LOG(m_logger) << "checkAndUpdateReceiveCounter: " << a_counter << ", " << m_ReceiveCounter; if (std::numeric_limits::max() == ++m_ReceiveCounter) m_ReceiveCounter = 0; if (m_ReceiveCounter != a_counter) { if (m_ReceiveCounter > a_counter) { ERROR_LOG(m_logger) << "It seems, the telegram with counter " << a_counter << " was resent from " << getName() << "! Expected: " << m_ReceiveCounter; } else { ERROR_LOG(m_logger) << "It seems, we lost one or several telegrams from " << getName() << ": receive-counter of received telegram: " << a_counter << "; Expected: " << m_ReceiveCounter; } #if 0 m_ReceiveCounter--; return false; #else m_ReceiveCounter = a_counter; return true; #endif } return true; } void PrtCnvCentralThread::timeoutTick() { if (m_DoCleanup) return; if (PrtCnvSocketThread::e_WORKING != m_State) return; try { N_CoPrtCnv::CoPrtCnvCentralTelegram *centralInjectTelegram; while ((centralInjectTelegram = m_InjectionThread->getNextCentralInjectionTelegram())) { sendTelegram(centralInjectTelegram); delete centralInjectTelegram; } } catch (std::exception& e) { ERROR_LOG(m_logger) << "Catched an unexpected exception: " << e.what() << " => could not inject radio-telegram, but we proceed..."; } sendAck(); } void PrtCnvCentralThread::timeoutAliveAndConnection() { INFO_LOG(m_logger) << "timeoutAliveAndConnection socketState=" << int(m_webSocket ? m_webSocket->state() : -1) << " m_State=" << m_State << " m_Socket=" << m_webSocket; switch (m_State) { case PrtCnvSocketThread::e_CONNECTING: connectSocket(); break; case PrtCnvSocketThread::e_WORKING: if (!m_bDirectConnection) { disconnectSocket(); m_reconnectTimer->start(); } break; case PrtCnvSocketThread::e_DISCONNECTING: break; default: ERROR_LOG(m_logger) << "timeoutAlive for m_State=" << m_State << " not implemented"; } } void PrtCnvCentralThread::sendAck() { TrainingAreaBase::TetraIcd::Acknowledge ack; ack.timeStamp = time(nullptr) * 1000; ack.ackCounter = m_ReceiveAckCounter; N_CoPrtCnv::CoPrtCnvCentralTelegram tele(ack); if (m_webSocket) m_webSocket->sendTextMessage(QString::fromUtf8(tele.data())); else emit directTextMessage(tele.data()); TRACE_LOG(m_logger) << "Wrote ack: " << tele.data(); } void PrtCnvCentralThread::sendTelegram(N_CoPrtCnv::CoPrtCnvCentralTelegram *a_telegram) { PrtCnvStoredCentralTelegram *handle = nullptr; sendTelegram(a_telegram, handle); } void PrtCnvCentralThread::sendTelegram(N_CoPrtCnv::CoPrtCnvCentralTelegram *a_telegram, PrtCnvStoredCentralTelegram *a_handle) { using namespace TrainingAreaBase::TetraIcd; if (a_telegram->telegramId() == TelegramId::DEAKTIVIERUNG_EINZELMINE) { const MineTriggered tele(MineTriggered::fromJson(a_telegram->jsonData())); m_alreadyTriggeredMines[tele.mineAreaId].insert(tele.mineIndex - 1); INFO_LOG(m_logger) << "Adding triggered mine with id " << tele.mineIndex << " to area id " << tele.mineAreaId; } const int writeCounter = a_handle ? a_handle->writeCounter() : 0; if (writeCounter != 0) a_telegram->setCounter(writeCounter); else a_telegram->setCounter(++m_SendCounter); DEBUG_LOG(m_logger) << "Send telegram to central: telId=" << a_telegram->telegramId() << " wcounter=" << a_telegram->counter(); if (m_webSocket) m_webSocket->sendTextMessage(QString::fromUtf8(a_telegram->data())); else emit directTextMessage(a_telegram->data()); if (a_handle) m_DbThread->centralTelegramSent(a_telegram->counter(), a_handle); m_LogThread->writeCentralTelegramLog(a_telegram, true, 1); } void PrtCnvCentralThread::handleAnforderungStatistikBroadcastNachforderungen() { const auto nachforderungsStat = m_DbThread->getClaimStatistics(); TrainingAreaBase::TetraIcd::BroadcastStatistics tele; tele.header.equipmentNumber = m_VirtualTeilnehmerId; tele.header.timeStamp = time(nullptr) * 1000LL; tele.information.reserve(nachforderungsStat.size()); for (const auto &it : nachforderungsStat) { TrainingAreaBase::TetraIcd::BroadcastInformation bi; bi.equipmentNumber = it.first; bi.broadcastRequests = it.second; tele.information.push_back(std::move(bi)); } N_CoPrtCnv::CoPrtCnvCentralTelegram telegram(tele); sendTelegram(&telegram); } void PrtCnvCentralThread::checkSend() { if (m_DoCleanup) return; if (PrtCnvSocketThread::e_WORKING != m_State) return; PrtCnvStoredCentralTelegram *handle = nullptr; auto *centralTelegram = m_DbThread->getNextCentralTelegram(handle); if (centralTelegram) { sendTelegram(centralTelegram, handle); delete centralTelegram; emit checkSendToCentral(); } } void PrtCnvCentralThread::handleAck(N_CoPrtCnv::CoPrtCnvCentralTelegram *a_telegram) { TrainingAreaBase::TetraIcd::Acknowledge ack(TrainingAreaBase::TetraIcd::Acknowledge::fromJson(a_telegram->jsonData())); m_DbThread->ackCentralTelegrams(ack.ackCounter); } void PrtCnvCentralThread::permanentStored(int32_t a_centralCounter) { if (m_DoCleanup) return; if (PrtCnvSocketThread::e_WORKING != m_State) return; DEBUG_LOG(m_logger) << "got permanentStored a_centralCounter=" << a_centralCounter; if (0 != a_centralCounter && a_centralCounter > m_ReceiveAckCounter) { m_ReceiveAckCounter = a_centralCounter; sendAck(); } } void PrtCnvCentralThread::onTetraConnectionState(bool bConnected) { TrainingAreaBase::TetraIcd::ProxyConnectionState connState; connState.connectedToSystem = bConnected; N_CoPrtCnv::CoPrtCnvCentralTelegram telegram(connState); sendTelegram(&telegram); } void PrtCnvCentralThread::onPlayerLicencedChanged(int equipNo, N_CoPrtCnv::E_LicensingKind kind) { TrainingAreaBase::TetraIcd::ParticipantLicenseChange lc; lc.equipmentNumber = equipNo; lc.licenseState = TrainingAreaBase::TetraIcd::ParticipantLicenseState::Unknown; switch (kind) { case N_CoPrtCnv::E_LicensingKind::Success: lc.licenseState = TrainingAreaBase::TetraIcd::ParticipantLicenseState::Licensed; break; case N_CoPrtCnv::E_LicensingKind::Rejected: lc.licenseState = TrainingAreaBase::TetraIcd::ParticipantLicenseState::Rejected; break; case N_CoPrtCnv::E_LicensingKind::Unlicense: lc.licenseState = TrainingAreaBase::TetraIcd::ParticipantLicenseState::Unlicense; break; } Q_ASSERT(lc.licenseState != TrainingAreaBase::TetraIcd::ParticipantLicenseState::Unknown); N_CoPrtCnv::CoPrtCnvCentralTelegram telegram(lc); sendTelegram(&telegram); } void PrtCnvCentralThread::onAreaEffectSimulationDeactivated(int wirkfeldKennung) { m_deactivatedAreaEffectSimulations.insert(wirkfeldKennung); m_alreadyTriggeredMines.remove(wirkfeldKennung); } void PrtCnvCentralThread::onAreaEffectSimulationActivated(int wirkfeldKennung) { m_deactivatedAreaEffectSimulations.remove(wirkfeldKennung); } void PrtCnvCentralThread::onRadioThreadInitDone() { m_bRadioThreadInitDone = true; if (!m_sendAckTimer->isActive()) m_sendAckTimer->start(); } bool PrtCnvCentralThread::handleSpecialTelegram(N_CoPrtCnv::CoPrtCnvCentralTelegram *a_telegram) { using namespace TrainingAreaBase::TetraIcd; switch (a_telegram->telegramId()) { case Constraints::id(): { ExerciseStateData tele(ExerciseStateData::fromJson(a_telegram->jsonData())); switch(tele.exerciseState) { case ExerciseStateData::ExerciseState::Terminated: { INFO_LOG(m_logger) << "Got Betriebszustandsmeldung Uebung_Aufraeumen => disconnect, cleanup and reconnect"; m_DbThread->setMachineState(1); permanentStored(a_telegram->counter()); m_DoCleanup = true; m_CentralStopped = false; m_RadioStopped = false; m_sendAckTimer->stop(); m_DbThread->waitUntilDbWritten(); disconnectSocket(); m_RadioThread->prepareCleanUp(); emit stopRadio(); m_reconnectTimer->start(); break; } case ExerciseStateData::ExerciseState::Started: m_DbThread->setPause(false); m_RadioThread->sendEnvironmentTelegramTimerNow(); break; case ExerciseStateData::ExerciseState::Paused: m_DbThread->setPause(true); m_RadioThread->sendEnvironmentTelegramTimerNow(); break; default: break; } return false; } case Constraints::id(): m_DbThread->centralCounterReadyForAck(a_telegram->counter()); handleAnforderungStatistikBroadcastNachforderungen(); return false; case Constraints::id(): { const ExerciseEquipmentAssignmentComplete tele(ExerciseEquipmentAssignmentComplete::fromJson(a_telegram->jsonData())); m_RadioThread->exerciseEquipmentChanged(tele); return false; } default: break; } return true; } void PrtCnvCentralThread::connectSocket() { const QString host = m_Config->getString(getName(), QStringLiteral("HostName"), QStringLiteral("EXCTDS01")); if (host == QLatin1String("directConnection")) { m_bDirectConnection = true; connected(); return; } m_bDirectConnection = false; disconnectSocket(); m_webSocket = new QWebSocket; connect(m_webSocket, &QWebSocket::stateChanged, this, &PrtCnvCentralThread::stateChanged); connect(m_webSocket, &QWebSocket::textMessageReceived, this, &PrtCnvCentralThread::onTextMessageReceived); connect(m_webSocket, &QWebSocket::connected, this, &PrtCnvCentralThread::connected); connect(m_webSocket, &QWebSocket::disconnected, this, &PrtCnvCentralThread::disconnected); const uint16_t port = static_cast(m_Config->getInteger(getName(), QStringLiteral("Port"), 4934)); const QUrl scmUrl(QStringLiteral("ws://%1:%2").arg(host).arg(port)); m_webSocket->open(scmUrl); m_State = PrtCnvSocketThread::e_CONNECTING; } void PrtCnvCentralThread::disconnectSocket() { m_State = PrtCnvSocketThread::e_INITIAL; if (!m_webSocket) return; m_State = PrtCnvSocketThread::e_DISCONNECTING; m_webSocket->disconnect(this); m_webSocket->close(); m_webSocket->deleteLater(); m_webSocket = nullptr; } }