Files
qt5-legatus-legacy/TetraIF/PrtCnv/PrtCnvDbThread.cpp
2026-08-13 18:33:10 +02:00

1656 lines
60 KiB
C++

// **************************************************************************
//
// (C) Copyright 2009 Swiss SIMTEC AG
//
// **************************************************************************
//
// $HeadURL$
// $LastChangedBy$
// $LastChangedDate$
// $LastChangedRevision$
//
// Implementation of the Class PrtCnvDbThread
// **************************************************************************
#include "PrtCnvDbThread.hpp"
#include "PrtCnvThreadStarter.hpp"
#include "PrtCnvCentralThread.hpp"
#include "PrtCnvMember.hpp"
#include "PrtCnvRadioThread.hpp"
#include "PrtCnvStoredCentralTelegram.hpp"
#include "PrtCnvStoredRadioTelegram.hpp"
#include <CoPrtCnv/CoPrtCnvRadioTelegram.hpp>
#include <CoPrtCnv/CoPrtCnvTelegramConverter.hpp>
#include <CoPrtCnv/CoPrtCnvCentralTelegram.hpp>
#include <CoPrtCnv/CoPrtCnvHexDumpHelper.hpp>
#include <QtCore/QCoreApplication>
#include <QtCore/QDir>
#include <ngldb/dbtransaction.h>
#include <ngldb/nsqlquery.h>
#include <QtSql/QSqlError>
#include <iostream>
namespace PrtCnv
{
// ---------------------------------------------------------------
// Constructor with thread-starter as argument
// ---------------------------------------------------------------
PrtCnvDbThread::PrtCnvDbThread(const QString &dbDriver,
const QString &dbHostName,
const QString &dbDatabaseName,
const QString &schemaName,
const QString &dbUserName,
const QString &dbPassword,
PrtCnv::PrtCnvThreadStarter *a_parent)
: PrtCnvThread(a_parent, "PrtCnvDbThread")
, m_BattleFieldOpStateMutex(QMutex::Recursive)
, m_MemberListByIdMutex(QMutex::Recursive)
, m_CentralListMutex(QMutex::Recursive)
, m_OldCentralAckCounter(0)
, m_RadioListMutex(QMutex::Recursive)
, m_SqlCmdListMutex(QMutex::Recursive)
, m_BroadcastMapMutex(QMutex::Recursive)
, m_DoClean(false)
, m_DoWaitUntilSqlWritten(false)
, m_DoStorePermanent(false)
, m_ResetCounter(0)
, m_BattleFieldOperatingState(0)
, m_EnvironmentTelegram(nullptr)
, m_DoProceedCleanup(false)
, m_OldCentralCounter(0)
, m_StorePermanentPending(0)
, m_NbrOfAttempts(5)
, m_dbDriver(dbDriver.isEmpty() ? QStringLiteral("QSQLITE") : dbDriver)
, m_dbHostName(dbHostName)
, m_dbDatabaseName(dbDatabaseName)
, m_dbSchemaName(schemaName)
, m_dbUserName(dbUserName)
, m_dbPassword(dbPassword)
{
m_Name = QStringLiteral("Db");
}
PrtCnvDbThread::~PrtCnvDbThread()
{
closeDb();
qDeleteAll(m_BroadcastList);
qDeleteAll(m_RadioList);
qDeleteAll(m_CentralList);
for (const auto &p : m_MemberListById)
delete p.second;
}
// ---------------------------------------------------------------
// reinit database
// ---------------------------------------------------------------
static const int s_iCurrentDatabaseSchemeVersion = 1;
void PrtCnvDbThread::reinitDatabase()
{
DBTransaction trans(m_db);
NSqlQuery q(m_db);
if (m_db.driverName() == QStringLiteral("QSQLITE"))
{
q.exec(QStringLiteral("DROP TABLE IF EXISTS %1;").arg(versionTableName()));
q.exec(QStringLiteral("CREATE TABLE %1 ("
"version integer,"
"date text,"
"product text,"
"change text)").arg(versionTableName()));
q.exec(QStringLiteral("INSERT INTO %1 (version, date, product, change) VALUES "
"(1, '2012-11-20', 'TetraInterface', 'Initial version table');").arg(versionTableName()));
q.exec(QStringLiteral("DROP TABLE IF EXISTS %1;").arg(toCentralTableName()));
q.exec(QStringLiteral("CREATE TABLE %1 ("
"id INTEGER PRIMARY KEY AUTOINCREMENT,"
"tel_id INTEGER,"
"counter INTEGER DEFAULT 0,"
"data BLOB);").arg(toCentralTableName()));
q.exec(QStringLiteral("CREATE INDEX counterIndex ON %1(counter);").arg(toCentralTableName()));
q.exec(QStringLiteral("DROP TABLE IF EXISTS %1;").arg(toRadioTableName()));
q.exec(QStringLiteral("CREATE TABLE %1 ("
"id INTEGER PRIMARY KEY AUTOINCREMENT,"
"tel_id INTEGER,"
"counter INTEGER DEFAULT 0,"
"broadcast_index INTEGER DEFAULT 0,"
"reset_counter INTEGER DEFAULT 0,"
"data BLOB);").arg(toRadioTableName()));
q.exec(QStringLiteral("CREATE INDEX bcindex ON %1(reset_counter, broadcast_index);").arg(toRadioTableName()));
q.exec(QStringLiteral("DROP TABLE IF EXISTS %1;").arg(memberTableName()));
q.exec(QStringLiteral("CREATE TABLE %1 ("
"id INTEGER PRIMARY KEY,"
"issi INTEGER,"
"licensed INTEGER DEFAULT 0,"
"claims INTEGER DEFAULT 0,"
"exercise_id INTEGER"
");").arg(memberTableName()));
q.exec(QStringLiteral("DROP TABLE IF EXISTS %1;").arg(statusTableName()));
q.exec(QStringLiteral("CREATE TABLE %1 ("
"id INTEGER PRIMARY KEY,"
"reset_counter INTEGER,"
"ctas_operating_state INTEGER,"
"machine_state INTEGER" /* 0=>normal; 1=>DB-cleanup*/
");").arg(statusTableName()));
}
q.exec(QStringLiteral("DELETE FROM %1 WHERE id = 1;").arg(statusTableName()));
q.exec(QStringLiteral("INSERT INTO %1 "
"(id, reset_counter, ctas_operating_state, machine_state) "
"VALUES (1, %2, %3, 0);")
.arg(statusTableName()).arg(int(m_ResetCounter)).arg(int(m_BattleFieldOperatingState)));
q.exec(QStringLiteral("DELETE FROM %1 WHERE id = 0;").arg(memberTableName()));
trans.commit();
}
// ---------------------------------------------------------------
// creates or restores the permanent database
// ---------------------------------------------------------------
bool PrtCnvDbThread::createDatabase()
{
closeDb();
m_db = QSqlDatabase::addDatabase(m_dbDriver, QStringLiteral("PrtCnv"));
if (m_dbDriver == QLatin1String("QSQLITE"))
{
QString path = m_Config->getString(getName(), QStringLiteral("DatabaseFileName"), QStringLiteral("telegrams.db"));
if (!QDir::isAbsolutePath(path))
{
path = QCoreApplication::applicationDirPath() + QLatin1String("/../") + path;
QFileInfo fi(path);
QDir().mkpath(fi.absolutePath());
}
m_db.setDatabaseName(path);
INFO_LOG(m_logger) << "Use db-file: " << path;
std::cout << "Use db-file: " << qPrintable(path) << std::endl;
}
else
{
m_db.setDatabaseName(m_dbDatabaseName);
m_db.setHostName(m_dbHostName);
m_db.setUserName(m_dbUserName);
m_db.setPassword(m_dbPassword);
}
DEBUG_LOG(m_logger) << "Connection name: " << m_db.connectionName();
if (!m_db.open())
{
FATAL_LOG(m_logger) << "Error while opening database: " << m_db.lastError().text();
return false;
}
if (m_DoCleanDb)
{
NOTICE_LOG(m_logger) << "Permanent data will be erased...";
reinitDatabase();
}
else
{
NOTICE_LOG(m_logger) << "Check database...";
NSqlQuery q(m_db);
if (m_db.driverName() == QStringLiteral("QSQLITE"))
{
if (!q.execAndFetch(QStringLiteral("SELECT MAX(version) FROM %1;").arg(versionTableName())))
{
ERROR_LOG(m_logger) << "Could not read version table - reinitialize database";
reinitDatabase();
}
else
{
// check for correct version see version table in reinitDatabase()!
int iVersion = q.value(0).toInt();
if (iVersion != s_iCurrentDatabaseSchemeVersion)
{
ERROR_LOG(m_logger) << "Database version does not match (is: "
<< iVersion << " expected: " << s_iCurrentDatabaseSchemeVersion
<< " - reinitialize database";
reinitDatabase();
}
}
}
if (!q.exec(QStringLiteral("SELECT * from %1 LIMIT 1;").arg(statusTableName())))
{
ERROR_LOG(m_logger) << "Failure while checking DB-table Status";
std::cout << "Failure while checking DB-table Status. Start PrtCnv with --cleandb" << std::endl;
return false;
}
if (!q.exec(QStringLiteral("SELECT * from %1 LIMIT 1;").arg(toCentralTableName())))
{
ERROR_LOG(m_logger) << "Failure while checking DB-table ToCentral";
std::cout << "Failure while checking DB-table ToCentral. Start PrtCnv with --cleandb" << std::endl;
return false;
}
if (!q.exec(QStringLiteral("SELECT * from %1 LIMIT 1;").arg(toRadioTableName())))
{
ERROR_LOG(m_logger) << "Failure while checking DB-table ToRadio";
std::cout << "Failure while checking DB-table ToRadio. Start PrtCnv with --cleandb" << std::endl;
return false;
}
// check whether the broadcast-member is ok
if (!q.exec(QStringLiteral("SELECT issi FROM %1 WHERE id=0;").arg(memberTableName())))
{
ERROR_LOG(m_logger) << "Failure while checking DB-table Member";
std::cout << "Failure while checking DB-table Member. Start PrtCnv with --cleandb" << std::endl;
return false;
}
const int broadcastIssi = m_Config->getInteger(QStringLiteral("Radio"), QStringLiteral("BroadcastISSI"), 5000);
if (q.next())
{
const int dbBroadcastIssi = q.value(0).toInt();
if (broadcastIssi != dbBroadcastIssi)
{
WARNING_LOG(m_logger) << "Broadcast-ISSI mismatch (DB: " << dbBroadcastIssi << "; Configfile: " << broadcastIssi << ")";
q.exec(QStringLiteral("DELETE FROM %1 WHERE id = 0;").arg(memberTableName()));
licenseMember(addMember(0, broadcastIssi));
}
}
else
{
licenseMember(addMember(0, broadcastIssi));
}
INFO_LOG(m_logger) << "Restore permanent data (after crash)...";
m_DoStorePermanent = false;
INFO_LOG(m_logger) << "restoreStatus";
// restore important varibles
if (!q.execAndFetch(QStringLiteral("SELECT reset_counter, ctas_operating_state, machine_state FROM %1;").arg(statusTableName())))
{
ERROR_LOG(m_logger) << "Error reading Status from DB";
std::cout << "Error reading Status from DB. Start PrtCnv with --cleandb" << std::endl;
return false;
}
m_ResetCounter = q.value(0).toInt();
INFO_LOG(m_logger) << "Restore resetCounter=" << (int)m_ResetCounter;
// check whether the crash happend just while cleaning up DB
if (1 == q.value(2).toInt())
{
INFO_LOG(m_logger) << "Crash happenend while cleaning-DB";
// crash while cleaning DB => clean-DB and restart
m_DoStorePermanent = true;
m_DoProceedCleanup = true;
return true;
}
m_BattleFieldOperatingState = q.value(1).toInt();
INFO_LOG(m_logger) << "Restore battleFieldOperatingState=" << (int)m_BattleFieldOperatingState;
// restore the members
if (!q.exec(QStringLiteral("SELECT id, issi, licensed, claims, exercise_id FROM %1;").arg(memberTableName())))
{
ERROR_LOG(m_logger) << "Error reading Member from DB";
std::cout << "Error reading Member from DB. Start PrtCnv with --cleandb" << std::endl;
return false;
}
while (q.next())
{
int id = q.value(0).toInt();
int issi = q.value(1).toInt();
bool licensed = q.value(2).toBool();
int claims = q.value(3).toInt();
int exerciseId = q.value(4).toInt();
INFO_LOG(m_logger) << "Restore member: id=" << id << " issi=" << issi << " licensed=" << licensed
<< " claims=" << claims << " exerciseId " << exerciseId;
PrtCnvMember *member = addMember(id, issi);
if (member)
{
member->m_ClaimCount = claims;
member->m_loggedInExerciseId = exerciseId;
if (licensed)
{
bool ok = licenseMember(member);
if (ok)
{
//emit signal
emit playerLicencedChanged(member->m_Id, N_CoPrtCnv::E_LicensingKind::Success);
}
}
}
}
// restore radio-telegrams
// Sortierung nach resetCounter + broadcastindex weil die Daten dann in
// getAllFinishedBroadcastsFromDb() der Reihenfolge nach abholt werden
// also müssen wir sie hier richtig herum in m_RadioList eintragen
if (!q.exec(QStringLiteral("SELECT id, counter, broadcast_index, data "
"FROM %1 ORDER BY reset_counter, broadcast_index, id;").arg(toRadioTableName())))
{
ERROR_LOG(m_logger) << "Error reading from ToRadio table";
std::cout << "Error reading from ToRadio table. Start PrtCnv with --cleandb" << std::endl;
return false;
}
while (q.next())
{
const qint64 id = q.value(0).toLongLong();
const int counter = q.value(1).toInt();
const int bcIndex = q.value(2).toInt();
const QByteArray telData(q.value(3).toByteArray());
auto radioTelegram(std::make_shared<N_CoPrtCnv::CoPrtCnvRadioTelegram>(telData));
PrtCnvStoredRadioTelegram *storedTelegram =
new PrtCnvStoredRadioTelegram(radioTelegram);
storedTelegram->m_Id = id;
// --------------------------
// ----------------------------
if (0 == counter)
{
// telegram not yet sent to radio
m_RadioList.push_back(storedTelegram);
INFO_LOG(m_logger) << "Restore unsent radiotelegram: id=" << storedTelegram->m_Id
<< " telId=" << storedTelegram->m_RadioTelegram->getTelID()
<< " BC=" << storedTelegram->m_RadioTelegram->getBroadcastIndex()
<< " counter=" << counter;
}
else
{
// telegram sent to radio
if (bcIndex == 0)
{
if (radioTelegram->getTelID() == GEF_UMGEBUNGS_MELDUNG2)
m_RadioList.push_back(storedTelegram);
else
delete storedTelegram; // normally not possible to get here, but just in case
}
else
{
// => broadcast-telegram
radioTelegram->setBroadcastIndex(bcIndex);
bool dummy;
radioTelegram->mustBeSentAsBroadcast(0xFFFFFFFF, dummy);
radioTelegram->mustBeSentAsBroadcast(0xFFFFFFFF, dummy);
radioTelegram->mustBeSentAsBroadcast(0xFFFFFFFF, dummy);
m_BroadcastList.push_back(storedTelegram);
INFO_LOG(m_logger) << "Restore broadcast-radiotelegram: id="
<< storedTelegram->m_Id << " telId=" << storedTelegram->m_RadioTelegram->getTelID()
<< " BC=" << storedTelegram->m_RadioTelegram->getBroadcastIndex()
<< " counter=" << counter;
}
}
}
// restore central-telegrams
q.exec(QStringLiteral("UPDATE %1 SET counter=0;").arg(toCentralTableName())); // => resend all not acked telegrams
if (!q.exec(QStringLiteral("SELECT id, data FROM %1 ORDER BY id;").arg(toCentralTableName())))
{
ERROR_LOG(m_logger) << "Error reading from ToCentral table";
std::cout << "Error reading from ToCentral table. Start PrtCnv with --cleandb" << std::endl;
return false;
}
while (q.next())
{
const qint64 id = q.value(0).toLongLong();
const QByteArray telData(q.value(1).toByteArray());
N_CoPrtCnv::CoPrtCnvRadioTelegram radioTelegram(telData);
std::shared_ptr<N_CoPrtCnv::CoPrtCnvCentralTelegram>
centralTelegram(N_CoPrtCnv::CoPrtCnvTelegramConverter::convertToCentral(&radioTelegram));
if (centralTelegram)
{
PrtCnvStoredCentralTelegram *storedTelegram = new PrtCnvStoredCentralTelegram(centralTelegram, nullptr);
storedTelegram->m_Id = id;
m_CentralList.push_back(storedTelegram);
INFO_LOG(m_logger) << "Restore unsent or unacked centraltelegram: id=" << storedTelegram->m_Id
<< " telId=" << storedTelegram->m_CentralTelegram->telegramId()
<< " counter=" << storedTelegram->m_WriteCounter
<< " size=" << telData.size();
//NGL::NGLLoggerHelper::hexDumpHelper(m_logger, NGL::LogLevel::Info, telData, blobSize);
}
else
ERROR_LOG(m_logger) << "Internal error: conversion of telegram with telId=" << radioTelegram.getTelID() << " from radio to central failed";
}
if (m_db.driverName() == QLatin1String("QSQLITE"))
{
q.exec(QStringLiteral("VACUUM;"));
q.exec(QStringLiteral("PRAGMA journal_mode=WAL;"));
// Sync mode to NORMAL instead FULL - small speedup at the cost
// of a little higher risk for data loss on power failure
q.exec(QStringLiteral("PRAGMA synchronous=NORMAL;"));
}
}
m_DoStorePermanent = true;
return true;
}
// ---------------------------------------------------------------
// creates this thread and its childs
// ---------------------------------------------------------------
int32_t PrtCnvDbThread::create(NGL::Settings *settings)
{
INFO_LOG(m_logger) << "create";
PrtCnvThread::create(settings);
m_NbrOfAttempts = m_Config->getInteger(QStringLiteral("Central"), QStringLiteral("NbrOfAttempts"), 5);
if (!createDatabase())
{
ERROR_LOG(m_logger) << "Trying to create clean database after error";
m_DoCleanDb = true;
if (!createDatabase())
exit(-2);
}
connect(this, &PrtCnvDbThread::doStorePermanent, this, &PrtCnvDbThread::storePermanent, Qt::QueuedConnection);
connect(this, &PrtCnvDbThread::doClean, this, &PrtCnvDbThread::clean, Qt::QueuedConnection);
PrtCnvCentralThread *centralThread = qobject_cast<PrtCnvCentralThread *>(m_Parent->getOtherThread(PrtCnvApplication::e_Central));
connect(this, &PrtCnvDbThread::permanentStored, centralThread, &PrtCnvCentralThread::permanentStored, Qt::QueuedConnection);
return 0;
}
// ---------------------------------------------------------------
// initializes this thread and its childs
// ---------------------------------------------------------------
int32_t PrtCnvDbThread::init()
{
INFO_LOG(m_logger) << "init";
PrtCnvThread::init();
m_CentralAckCounter = 0;
m_OldCentralAckCounter = 0;
m_OldCentralCounter = 0;
m_DoClean = false;
m_DoWaitUntilSqlWritten = false;
return 0;
}
// ---------------------------------------------------------------
// starts this thread and its childs
// ---------------------------------------------------------------
int32_t PrtCnvDbThread::start()
{
INFO_LOG(m_logger) << "start";
PrtCnvThread::start();
return 0;
}
// ---------------------------------------------------------------
// stops this thread and its childs
// ---------------------------------------------------------------
int32_t PrtCnvDbThread::stop()
{
PrtCnvThread::stop();
m_db.close();
m_db = QSqlDatabase();
return 0;
}
// ---------------------------------------------------------------
// stores the telegram to be sent to central in working-list
// and also in Db. Because storing data in XDR_Wrapper_it-
// format is complicated (has pointer to data), we store them
// in the radio-telegram-format).
// Attention: Be aware that centralTelegram and radioTelegram
// contain the same data.
// The DbThread takes control over centralTelegram and
// radioTelegram and deletes them when no longer needed.
// ---------------------------------------------------------------
void PrtCnvDbThread::storeCentralTelegram(const std::shared_ptr<N_CoPrtCnv::CoPrtCnvCentralTelegram> &a_centralTelegram,
const N_CoPrtCnv::CoPrtCnvRadioTelegram &a_radioTelegram)
{
PrtCnvStoredCentralTelegram *storedTelegram =
new PrtCnvStoredCentralTelegram(a_centralTelegram,
std::make_shared<N_CoPrtCnv::CoPrtCnvRadioTelegram>(a_radioTelegram));
QMutexLocker l(&m_CentralListMutex);
m_CentralList.push_back(storedTelegram);
// INFO_LOG(m_logger) << "Store Central-Telegram with telId=" << a_telegram->getTelID() << " and rc=" << a_telegram->zaehler();
sendStorePermanentSignal();
}
// ---------------------------------------------------------------
// returns the next telegram to send to central.
// returns NULL if no telegram found.
// ATTENTION: The caller is responsible to deallocate the
// returned centralTelegram!
// ---------------------------------------------------------------
N_CoPrtCnv::CoPrtCnvCentralTelegram *PrtCnvDbThread::getNextCentralTelegram(PrtCnvStoredCentralTelegram *&a_handle)
{
// get next telegram to send to central from DB
QMutexLocker l(&m_CentralListMutex);
std::vector<PrtCnvStoredCentralTelegram *>::const_iterator iter = m_CentralList.begin();
std::vector<PrtCnvStoredCentralTelegram *>::const_iterator end = m_CentralList.end();
for (; iter != end; ++iter)
{
PrtCnvStoredCentralTelegram *storedTelegram = *iter;
if (!storedTelegram || !storedTelegram->m_CentralTelegram)
continue;
if (0 == storedTelegram->m_WriteCounter || storedTelegram->m_DoResend)
{
const auto &telegram = storedTelegram->m_CentralTelegram;
storedTelegram->m_DoResend = false;
a_handle = storedTelegram;
DEBUG_LOG(m_logger) << "Next telegram to send to central: id=" << storedTelegram->m_Id
<< " a_telId=" << telegram->telegramId();
// allocate a central-telegram
return new N_CoPrtCnv::CoPrtCnvCentralTelegram(*telegram);
}
}
return nullptr;
}
// ---------------------------------------------------------------
// Marks the central-telegram as to be sent
// ---------------------------------------------------------------
void PrtCnvDbThread::centralTelegramSent(const int32_t a_counter, PrtCnvStoredCentralTelegram *a_handle)
{
// update sendcounter
QMutexLocker l(&m_CentralListMutex);
a_handle->m_WriteCounter = a_counter;
a_handle->m_WriteCounterChanged = true;
sendStorePermanentSignal();
}
// ---------------------------------------------------------------
// Indicates the Db, which telegrams were successful received
// by central. This will delete this telegrams from Db.
// ---------------------------------------------------------------
void PrtCnvDbThread::ackCentralTelegrams(int32_t a_counter)
{
//DEBUG_LOG(m_logger) << "Got Ack from Central with wc=" << a_counter;
m_CentralAckCounter = a_counter;
sendStorePermanentSignal();
}
// ---------------------------------------------------------------
// Updates the counter for unsuccessful sending the telegram.
// If a maximal value of this counter is reached, the central-
// telegram will be deleted.
// a_handle could be deleted after calling this function!
// This function could block some milliseconds, because it
// must wait until all changes are written to DB.
// ---------------------------------------------------------------
void PrtCnvDbThread::centralTelegramCouldNotBeSent(PrtCnvStoredCentralTelegram *a_handle)
{
if (a_handle)
{
// delete the telegram if much enough unsuccessfully sent
QMutexLocker l(&m_CentralListMutex);
if (++(a_handle->m_Attempts) >= 5)
{
WARNING_LOG(m_logger) << "Delete unsendable Central-Telegram with id=" << a_handle->m_Id
<< " telId=" << a_handle->m_CentralTelegram->telegramId()
<< " rc=" << a_handle->m_CentralTelegram->counter() << " wc=" << a_handle->m_WriteCounter;
a_handle->m_DoDelete = true;
// mark in Db telegram as sent
}
l.unlock();
waitUntilDbWritten();
}
}
// ---------------------------------------------------------------
// Stores the telegram to be sent to radio in working-list and
// also in Db.
// The DbThread takes control over a_telegram and deletes it
// when no longer needed.
// ---------------------------------------------------------------
void PrtCnvDbThread::storeRadioTelegram(const std::shared_ptr<N_CoPrtCnv::CoPrtCnvRadioTelegram> &a_telegram,
const int32_t a_readCounter)
{
PrtCnvStoredRadioTelegram *storedTelegram = new PrtCnvStoredRadioTelegram(a_telegram);
storedTelegram->m_ReadCounter = a_readCounter;
{
QMutexLocker l(&m_RadioListMutex);
m_RadioList.push_back(storedTelegram);
}
if (a_telegram)
DEBUG_LOG(m_logger) << "Store Radio-Telegram with telId=" << a_telegram->getTelID()
<< " rc=" << a_readCounter
<< " wc=" << int(storedTelegram->m_WriteCounter);
if (m_DoStorePermanent)
sendStorePermanentSignal();
}
// ---------------------------------------------------------------
// This function causes acking a central-telegram, which must
// not be stored permanent in Db, because it will be handled
// immediately (i.e.
// AnforderungStatistikBroadcastNachforderungen).
// ---------------------------------------------------------------
void PrtCnvDbThread::centralCounterReadyForAck(const int32_t a_centralCounter)
{
storeRadioTelegram(nullptr, a_centralCounter);
DEBUG_LOG(m_logger) << "No Radio-Telegram to store, but rc=" << a_centralCounter << " must be acked";
}
// ---------------------------------------------------------------
// returns the next broadcast-telegram from list.
// Deletes the telegram from this list.
// returns NULL if list empty.
// ---------------------------------------------------------------
std::shared_ptr<N_CoPrtCnv::CoPrtCnvRadioTelegram> PrtCnvDbThread::getNextBroadcastTelegram()
{
QMutexLocker l(&m_BroadcastMapMutex);
if (m_BroadcastList.empty())
return nullptr;
PrtCnvStoredRadioTelegram *storedRadioTelegram = m_BroadcastList.front();
std::shared_ptr<N_CoPrtCnv::CoPrtCnvRadioTelegram> radioTelegram;
// check whether the telegram is already stored permanent in DB
if (storedRadioTelegram->m_Id != 0 && !storedRadioTelegram->m_WriteCounterChanged)
{
radioTelegram = storedRadioTelegram->m_RadioTelegram;
m_BroadcastList.pop_front();
delete storedRadioTelegram;
}
return radioTelegram;
}
// ---------------------------------------------------------------
// returns the next telegram to send to radio, respecting the
// priority.
// returns NULL if no radio-telegram found.
// ---------------------------------------------------------------
std::shared_ptr<N_CoPrtCnv::CoPrtCnvRadioTelegram> PrtCnvDbThread::getNextRadioTelegram(PrtCnvStoredRadioTelegram *&a_handle)
{
// get next telegram respecting the priority to send to radio from DB
std::shared_ptr<N_CoPrtCnv::CoPrtCnvRadioTelegram> telegram;
uint8_t priority = 255;
QMutexLocker l(&m_RadioListMutex);
for (PrtCnvStoredRadioTelegram *storedTelegram: m_RadioList)
{
const auto &radioTelegram = storedTelegram->m_RadioTelegram;
if (radioTelegram)
{
if (!storedTelegram->m_IsSent)
{
// search the telegram with the lowest priority
uint8_t newPriority = radioTelegram->getPriority();
if (newPriority < priority)
{
priority = newPriority;
telegram = radioTelegram;
a_handle = storedTelegram;
}
}
}
}
return telegram;
}
// ---------------------------------------------------------------
// Marks the radio-telegram as to be sent
// ---------------------------------------------------------------
void PrtCnvDbThread::radioTelegramSent(const uint32_t a_sendCounter, PrtCnvStoredRadioTelegram *a_handle, const uint32_t a_broadcastIndex)
{
QMutexLocker l(&m_RadioListMutex);
if (a_broadcastIndex)
{
DEBUG_LOG(m_logger) << "request update radio telegramm in db - sc: " << a_sendCounter
<< " telId: " << a_handle->m_RadioTelegram->getTelID()
<< " BC: " << a_broadcastIndex;
}
// update sendcounter in telegram
a_handle->m_WriteCounter = a_sendCounter;
a_handle->m_WriteCounterChanged = true;
a_handle->m_IsSent = true;
if (a_broadcastIndex != 0)
a_handle->m_BroadcastIndex = a_broadcastIndex;
if (a_handle->m_BroadcastIndex != 0)
{
// set the time for the 2nd broadcast
bool dummy;
a_handle->m_RadioTelegram->mustBeSentAsBroadcast(0xFFFFFFFF, dummy);
DEBUG_LOG(m_logger) << "Move Radio-Telegram with telId=" << a_handle->m_RadioTelegram->getTelID()
<< " and BC=" << a_handle->m_RadioTelegram->getBroadcastIndex()
<< " to BC-List of DB";
// if (a_handle->m_Id != 0)
// {
// M_DEBUG("Remove from radio-list")
// m_RadioList.remove(a_handle);
// }
QMutexLocker l2(&m_BroadcastMapMutex);
m_BroadcastList.push_back(a_handle);
}
else
{
// do not delete umgebungsmeldung, because we need it for restore after crash
if (a_handle->m_RadioTelegram->isNormalTelegram() && (a_handle->m_RadioTelegram->getTelID() == GEF_UMGEBUNGS_MELDUNG2))
{
// store only the last Umgebungsmeldung
if (m_EnvironmentTelegram)
m_EnvironmentTelegram->m_DoDelete = true;
m_EnvironmentTelegram = a_handle;
}
else
a_handle->m_DoDelete = true;
}
sendStorePermanentSignal();
}
void PrtCnvDbThread::deleteTelegramsFromDbById(const QString &table)
{
if (m_telegramsToDelete.empty())
return;
QStringList sl;
sl.reserve(m_telegramsToDelete.size());
for (const int64_t telId : m_telegramsToDelete)
sl += QString::number(telId);
NSqlQuery q(m_db);
q.exec(QStringLiteral("DELETE FROM %1 WHERE id IN (%2);").arg(table, sl.join(QLatin1Char(','))));
// cleanup
m_telegramsToDelete.clear();
}
void PrtCnvDbThread::updateTelegramsBroadcast(const QString &table)
{
if (m_telegramsToUpdateBC.empty())
return;
NSqlQuery q(m_db);
q.prepare(QStringLiteral("UPDATE %1 SET broadcast_index = :broadcast_index WHERE id = :id").arg(table));
for (const auto &p : m_telegramsToUpdateBC)
{
const qint64 telId = p.first;
const qint32 bcIndex = p.second;
q.bindValue(QStringLiteral(":broadcast_index"), bcIndex);
q.bindValue(QStringLiteral(":id"), telId);
q.exec();
DEBUG_LOG(m_logger) << "Update broadcastindex for id: " << telId << " to BC: " << bcIndex;
}
// cleanup
m_telegramsToUpdateBC.clear();
}
void PrtCnvDbThread::updateTelegramsWriteCounter(const QString &table)
{
if (m_telegramsToUpdateCounter.empty())
return;
NSqlQuery q(m_db);
q.prepare(QStringLiteral("UPDATE %1 SET counter = :counter WHERE id = :id").arg(table));
for (const auto &p : m_telegramsToUpdateCounter)
{
const qint64 telId = p.first;
const qint32 writeCounter = p.second;
q.bindValue(QStringLiteral(":counter"), writeCounter);
q.bindValue(QStringLiteral(":id"), telId);
q.exec();
DEBUG_LOG(m_logger) << "Update WriteCounter for id: " << telId << " to wc: " << writeCounter;
}
// cleanup
m_telegramsToUpdateCounter.clear();
}
/**
* Da SQL-Kommandos im Gegensatz zum Auslesen aus dem Socket zu langsam
* ist und wir hier recht lange Lock-Zeiten haben, müssen wir uns mit einer
* zweiteiligen Verarbeitung zufrieden geben
* 1. Step - Auswerten und Aktualiseren der Liste
* 2. Step - Ausführen der SQL-Kommandos
* Da wir an Tetra keine Bestätigung verschicken bzw. Tetra im Falle
* eines Crashes sowieso nicht alles puffern kann, sollte dies ok
* sein.
*/
void PrtCnvDbThread::storeCentralTelegrams()
{
QMutexLocker l(&m_CentralListMutex);
if (m_CentralList.empty())
return;
// DEBUG_LOG(m_logger) << "centralTelegrams.size=" << m_CentralList.size();
int iEmptyCount = 0;
for (auto &storedCentralTelegram : m_CentralList) // reference is important to be able to make them a nullptr in the vector
{
if (!storedCentralTelegram)
{
++iEmptyCount;
continue;
}
const auto centralTelegram = storedCentralTelegram->m_CentralTelegram;
if (0 == storedCentralTelegram->m_Id)
{
// new Telegram from Tetra, not yet in db -> insert it now
m_telegramsToInsert.emplace_back(std::make_pair(storedCentralTelegram, 0));
DEBUG_LOG(m_logger) << "Permanent Central-Telegram with id=" << storedCentralTelegram->m_Id
<< " data=" << centralTelegram->data()
<< " wc=" << storedCentralTelegram->m_WriteCounter;
}
else
{
// update counter for all sent central-telegrams
// ---------------------------------------------
const int centralAckCnt = m_CentralAckCounter;
if (storedCentralTelegram->m_WriteCounterChanged)
{
m_telegramsToUpdateCounter.emplace_back(std::make_pair(storedCentralTelegram->m_Id,
storedCentralTelegram->m_WriteCounter));
storedCentralTelegram->m_WriteCounterChanged = false;
DEBUG_LOG(m_logger) << "Update Central-Telegram with id=" << storedCentralTelegram->m_Id
<< " data=" << centralTelegram->data()
<< " wc=" << storedCentralTelegram->m_WriteCounter;
}
// delete all acked telegrams from central-list
// --------------------------------------------
else if ((storedCentralTelegram->m_WriteCounter > 0 &&
storedCentralTelegram->m_WriteCounter <= centralAckCnt) ||
storedCentralTelegram->m_DoDelete)
{
DEBUG_LOG(m_logger) << "Delete Central-Telegram with id=" << storedCentralTelegram->m_Id
<< " data=" << storedCentralTelegram->m_CentralTelegram->data()
<< " wc=" << storedCentralTelegram->m_WriteCounter;
// no need for single delete when centralAckCnt is higher - will be deleted later with where counter <= ackCounter
if (storedCentralTelegram->m_WriteCounter > centralAckCnt)
m_telegramsToDelete.push_back(storedCentralTelegram->m_Id);
// Removing from vector and DB occours some lines down
delete storedCentralTelegram;
storedCentralTelegram = nullptr;
}
}
}
// clean up m_CentralList if really needed
if (iEmptyCount > 50)
{
const auto isNull = [](const PrtCnvStoredCentralTelegram *t) { return t == nullptr; };
m_CentralList.erase(std::remove_if(m_CentralList.begin(), m_CentralList.end(), isNull),
m_CentralList.end());
}
l.unlock();
// DEBUG_LOG(m_logger) << "ready: " << m_CentralList.size();
// insert new telegrams into db
if (!m_telegramsToInsert.empty())
{
l.relock();
NSqlQuery q(m_db);
q.prepare(QStringLiteral("INSERT INTO %1 (tel_id, counter, data)"
"VALUES(:tel_id, :counter, :data);").arg(toCentralTableName()));
// TODO: ist das Locking mittels m_CentralListMutex überhaupt nötig?
// Wenn nein dann kann man dies hier etwas einfacher gestalten
for (const auto &p : m_telegramsToInsert)
{
PrtCnvStoredCentralTelegram *storedCentralTelegram = p.first;
storedCentralTelegram->m_WriteCounterChanged = false;
q.bindValue(QStringLiteral(":tel_id"), storedCentralTelegram->m_CentralTelegram->telegramId());
q.bindValue(QStringLiteral(":counter"), storedCentralTelegram->m_WriteCounter);
q.bindValue(QStringLiteral(":data"), storedCentralTelegram->m_RadioTelegram->data());
q.exec();
storedCentralTelegram->m_Id = q.lastInsertId().toInt();
}
l.unlock();
// DEBUG_LOG(m_logger) << "added new telegrams to db: " << m_telegramsToInsert.size();
m_telegramsToInsert.clear();
}
// update write counter for a telegramm
updateTelegramsWriteCounter(toCentralTableName());
// DEBUG_LOG(m_logger) << "updated write counter in db: " << m_telegramsToUpdateCounter.size();
// Jetzt alle Telegramme löschen die von der Zentrale bestätigt worden
// da m_CentralAckCounter von anderen Threads geändert werden kann, kann es
// sein dass schon mehr Telegramme aus der DB gelöscht werden als vorher
// aus m_CentralList. Dies wird aber im nächsten Aufruf wieder korrigiert
const unsigned int centralAckCnt = static_cast<unsigned int>(m_CentralAckCounter);
if (centralAckCnt > m_OldCentralAckCounter)
{
m_OldCentralAckCounter = centralAckCnt;
// wir arbeiten mit m_OldCentralAckCounter da sich m_CentralAckCounter
// währenddessen schon ändern kann
DEBUG_LOG(m_logger) << "Central delete all counter <= " << m_OldCentralAckCounter;
NSqlQuery q(m_db);
q.exec(QStringLiteral("DELETE FROM %1 WHERE counter > 0 AND counter <= %2;").arg(toCentralTableName()).arg(m_OldCentralAckCounter));
}
// remove all telegrams which were manually marked for deletion
deleteTelegramsFromDbById(toCentralTableName());
}
int32_t PrtCnvDbThread::storeRadioTelegrams()
{
QMutexLocker l(&m_RadioListMutex);
if (m_RadioList.empty())
return 0;
// M_DEBUG("Check for permanent actions on radio-list")
int32_t centralCounter = 0;
NSqlQuery q(m_db);
q.prepare(QStringLiteral("INSERT INTO %1 (tel_id, counter, broadcast_index, reset_counter, data) "
"VALUES (:tel_id, :counter, :broadcast_index, :reset_counter, :data);").arg(toRadioTableName()));
// DEBUG_LOG(m_logger) << "m_RadioList.size=" << m_RadioList.size();
std::list<PrtCnvStoredRadioTelegram *>::iterator radioIter = m_RadioList.begin();
while (radioIter != m_RadioList.end())
{
bool doDeleteThisItem = false;
PrtCnvStoredRadioTelegram *storedRadioTelegram = *radioIter;
const auto &radioTelegram = storedRadioTelegram->m_RadioTelegram;
if (radioTelegram)
{
DEBUG_LOG(m_logger) << "Handling Radio-Telegram with id=" << storedRadioTelegram->m_Id
<< " TelID: " << radioTelegram->getTelID()
<< " rc: " << int(radioTelegram->zaehler())
<< " BC: " << storedRadioTelegram->m_BroadcastIndex
<< " wc: " << int(storedRadioTelegram->m_WriteCounter);
// store radio-telegrams permanent
// -------------------------------
if (0 == storedRadioTelegram->m_Id)
{
if (storedRadioTelegram->m_ReadCounter > centralCounter)
centralCounter = storedRadioTelegram->m_ReadCounter;
storedRadioTelegram->m_WriteCounterChanged = false;
q.bindValue(QStringLiteral(":tel_id"), radioTelegram->getTelID());
q.bindValue(QStringLiteral(":counter"), storedRadioTelegram->m_WriteCounter);
q.bindValue(QStringLiteral(":broadcast_index"), storedRadioTelegram->m_BroadcastIndex);
q.bindValue(QStringLiteral(":reset_counter"), int(m_ResetCounter));
q.bindValue(QStringLiteral(":data"), radioTelegram->data());
q.exec();
storedRadioTelegram->m_Id = q.lastInsertId().toInt();
DEBUG_LOG(m_logger) << "Stored Telegramm in DB, id: " << storedRadioTelegram->m_Id;
// for broadcast-telegrams, we have 2 cases:
// 1. case: it will be written to DB, before it is sent to radio and then moved to broadcast-list of broadcast-manager
// 2. case: it will be sent to radio before it is written to DB. In this case we cannot remove it from m_RadioList until
// it is written to DB
// here we check whether case 2 has happenend
if (storedRadioTelegram->m_RadioTelegram->isMovedToBroadcastList())
{
dumpTelegram(m_logger,
radioTelegram->getBroadcastCounter(),
storedRadioTelegram->m_RadioTelegram.get());
DEBUG_LOG(m_logger) << "Removed Telegramm from radioList (place 1), id: " << storedRadioTelegram->m_Id;
doDeleteThisItem = true;
}
}
else
{
// delete all acked telegrams from radio-list
// ------------------------------------------
// do we have to delete the radio-telegram in the list and DB?
// we do only delete if the telegram is stored permanent in db and therefore the ack to central is sent
if (storedRadioTelegram->m_DoDelete)
{
DEBUG_LOG(m_logger) << "Delete Telegramm from DB, id: " << storedRadioTelegram->m_Id;
m_telegramsToDelete.push_back(storedRadioTelegram->m_Id);
// delete the telegram in the list
delete storedRadioTelegram;
doDeleteThisItem = true;
}
// update write-counter for all sent radio-telegrams
// -------------------------------------------------
else if (storedRadioTelegram->m_WriteCounterChanged)
{
m_telegramsToUpdateCounter.emplace_back(std::make_pair(storedRadioTelegram->m_Id,
storedRadioTelegram->m_WriteCounter));
if (storedRadioTelegram->m_BroadcastIndex)
m_telegramsToUpdateBC.emplace_back(std::make_pair(storedRadioTelegram->m_Id,
storedRadioTelegram->m_BroadcastIndex));
storedRadioTelegram->m_WriteCounterChanged = false;
DEBUG_LOG(m_logger) << "Update Write counter in DB, id: " << storedRadioTelegram->m_Id
<< " New wc: " << int(storedRadioTelegram->m_WriteCounter)
<< " new bc: " << storedRadioTelegram->m_BroadcastIndex;
if (storedRadioTelegram->m_RadioTelegram->isMovedToBroadcastList())
{
dumpTelegram(m_logger,
storedRadioTelegram->m_BroadcastIndex,
storedRadioTelegram->m_RadioTelegram.get());
DEBUG_LOG(m_logger) << "Removed Telegramm from radioList (place 2), id: " << storedRadioTelegram->m_Id;
doDeleteThisItem = true;
}
}
}
}
else
{
// telegrams which will not be stored, but acked (i.e. Anforderung Nachforderungs Statistik)
if (storedRadioTelegram->m_ReadCounter > centralCounter)
centralCounter = storedRadioTelegram->m_ReadCounter;
delete storedRadioTelegram;
doDeleteThisItem = true;
}
if (doDeleteThisItem)
radioIter = m_RadioList.erase(radioIter);
else
++radioIter;
}
l.unlock();
// update write counter for a telegramm
static const QString &table = toRadioTableName();
updateTelegramsWriteCounter(table);
// DEBUG_LOG(m_logger) << "updated write counter in db: " << m_telegramsToUpdateCounter.size();
// update BC for a telegramm
updateTelegramsBroadcast(table);
// remove all telegrams which were manually marked for deletion
deleteTelegramsFromDbById(table);
return centralCounter;
}
void PrtCnvDbThread::handleSqlCommandList()
{
// M_DEBUG("Check for permanent actions on sql-command-list")
std::vector<QString> sqlCommandList;
{
QMutexLocker l(&m_SqlCmdListMutex);
std::swap(m_SqlCommandList, sqlCommandList);
}
if (sqlCommandList.empty())
return;
NSqlQuery q(m_db);
for (const QString &sql : sqlCommandList)
{
DEBUG_LOG(m_logger) << "execute sql=" << sql;
q.exec(sql);
}
}
// ---------------------------------------------------------------
// Stores the radio-telegrams, central-telegrams in Db and
// executes the sql-command-list.
// ---------------------------------------------------------------
void PrtCnvDbThread::storePermanent()
{
m_StorePermanentPending = 0;
TRACE_LOG(m_logger) << "storePermanent m_DoClean=" << m_DoClean << " m_DoWaitUntilSqlWritten=" << m_DoWaitUntilSqlWritten;
if (m_DoClean)
{
m_DoWaitUntilSqlWritten = false;
return;
}
// generate the sql-commands on permanent DB
// -----------------------------------------
DBTransaction trans(m_db);
// permanent DB-actions for central
// ================================
storeCentralTelegrams();
// permanent DB-actions for radio
// ==============================
const int32_t centralCounter = storeRadioTelegrams();
// store the sql-commands (member)
// ======================
handleSqlCommandList();
// ready
trans.commit();
// M_DEBUG("inform central-thread")
if (0 != centralCounter && centralCounter != m_OldCentralCounter)
{
m_OldCentralCounter = centralCounter;
emit permanentStored(centralCounter);
}
m_DoWaitUntilSqlWritten = false;
// M_DEBUG("End check for permanent actions")
}
// ---------------------------------------------------------------
// Adds an sql-command in the sql-command-list. On next
// iteration of storePermanent() all command in list will be
// executed.
// ---------------------------------------------------------------
void PrtCnvDbThread::addSqlCommand(const QString &sql)
{
QMutexLocker l(&m_SqlCmdListMutex);
m_SqlCommandList.push_back(sql);
sendStorePermanentSignal();
}
// ---------------------------------------------------------------
// erases and initializes the database
// ---------------------------------------------------------------
void PrtCnvDbThread::cleanAndInit()
{
// emit a signal to clean, so we can be sure that all pending signals are worked
m_DoClean = true;
emit doClean();
// simply block until clean
while (m_DoClean)
QThread::sleep(1);
m_DoProceedCleanup = false;
}
// ---------------------------------------------------------------
// cleans up this thread and its childs
// ---------------------------------------------------------------
void PrtCnvDbThread::clean()
{
NSqlQuery q(m_db);
INFO_LOG(m_logger) << "Delete CentralList";
{
QMutexLocker l(&m_CentralListMutex);
for (PrtCnvStoredCentralTelegram *t : m_CentralList)
delete t;
m_CentralList.clear();
q.exec(QStringLiteral("DELETE FROM %1;").arg(toCentralTableName()));
}
m_OldCentralCounter = 0;
INFO_LOG(m_logger) << "Delete RadioList";
m_EnvironmentTelegram = nullptr;
{
QMutexLocker l(&m_RadioListMutex);
for (PrtCnvStoredRadioTelegram *t : m_RadioList)
delete t;
m_RadioList.clear();
q.exec(QStringLiteral("DELETE FROM %1;").arg(toRadioTableName()));
}
INFO_LOG(m_logger) << "Delete BroadcastList";
{
QMutexLocker l(&m_BroadcastMapMutex);
for (PrtCnvStoredRadioTelegram *t : m_BroadcastList)
delete t;
m_BroadcastList.clear();
}
// clean the members
// delete the member-maps
{
QMutexLocker l(&m_SqlCmdListMutex);
m_MemberListByISSI.clear();
{
QMutexLocker l2(&m_MemberListByIdMutex);
for (const auto &p : m_MemberListById)
delete p.second;
m_MemberListById.clear();
}
INFO_LOG(m_logger) << "Delete SqlList";
m_SqlCommandList.clear();
q.exec(QStringLiteral("DELETE FROM %1;").arg(memberTableName()));
}
// 'virtual' TNE for broadcasts
licenseMember(addMember(0, m_Config->getInteger(QStringLiteral("Radio"), QStringLiteral("BroadcastISSI"), 5000)));
m_DoClean = false;
}
// ---------------------------------------------------------------
// Checks whether central-telegrams have to be resent. If yes,
// all actions will be done, that getNextCentralTelegram()
// delivers the telegram(s) to be resent.
// Returns true if there must telegrams be resent.
// ---------------------------------------------------------------
bool PrtCnvDbThread::resendCentralTelegrams(const int32_t a_lastSendCounter)
{
const int centralAckCnt = m_CentralAckCounter;
if (a_lastSendCounter > centralAckCnt)
{
bool warnOnce = true;
QMutexLocker l(&m_CentralListMutex);
int iCount = 0;
std::vector<PrtCnvStoredCentralTelegram *>::const_iterator iter = m_CentralList.begin();
std::vector<PrtCnvStoredCentralTelegram *>::const_iterator end = m_CentralList.end();
for (; iter != end && iCount < 50; ++iter)
{
PrtCnvStoredCentralTelegram *storedTelegram = *iter;
if (!storedTelegram)
continue;
++iCount;
if (storedTelegram->m_WriteCounter > centralAckCnt)
{
if (warnOnce)
{
WARNING_LOG(m_logger) << "Resend central-telegrams from wc=" << a_lastSendCounter
<< " centralAckCounter=" << centralAckCnt;
warnOnce = false;
}
INFO_LOG(m_logger) << "Resend central-telegram with telId="
<< storedTelegram->m_CentralTelegram->telegramId()
<< " wc=" << storedTelegram->m_WriteCounter;
storedTelegram->m_DoResend = true;
}
}
return true;
}
return false;
}
// ---------------------------------------------------------------
// Prepares the telegrams not yet acked from central for
// reconnection. This will be done by setting the write-
// counter to 0.
// ---------------------------------------------------------------
void PrtCnvDbThread::prepareCentralTelegramsForReconnection()
{
INFO_LOG(m_logger) << "m_CentralAckCounter=" << m_CentralAckCounter;
waitUntilDbWritten();
INFO_LOG(m_logger) << "start";
m_CentralAckCounter = 0;
m_OldCentralAckCounter = 0;
m_OldCentralCounter = 0;
QMutexLocker l(&m_CentralListMutex);
for (PrtCnvStoredCentralTelegram *storedTelegram : m_CentralList)
{
if (!storedTelegram)
continue;
if (storedTelegram->m_WriteCounter != 0)
{
INFO_LOG(m_logger) << "***Prepare central-telegram with id=" << storedTelegram->m_Id
<< " telId=" << storedTelegram->m_CentralTelegram->telegramId()
<< " wc=" << storedTelegram->m_WriteCounter << " wcChg="
<< storedTelegram->m_WriteCounterChanged
<< " for resend after reconnection";
storedTelegram->m_WriteCounter = 0;
storedTelegram->m_WriteCounterChanged = false;
storedTelegram->m_DoResend = false;
}
}
INFO_LOG(m_logger) << "Ready for reconnection";
}
// ---------------------------------------------------------------
// Blocks until all SQL-commands were written to DB.
// ---------------------------------------------------------------
void PrtCnvDbThread::waitUntilDbWritten()
{
m_DoWaitUntilSqlWritten = true;
emit doStorePermanent();
while (m_DoWaitUntilSqlWritten)
QThread::usleep(10000); // 10 ms
// we have to do this twice, to be sure
m_DoWaitUntilSqlWritten = true;
emit doStorePermanent();
while (m_DoWaitUntilSqlWritten)
QThread::usleep(10000); // 10 ms
}
// ---------------------------------------------------------------
// adds a member in the 2 maps, sorted by id and ISSI
// returns the new member
// ---------------------------------------------------------------
PrtCnv::PrtCnvMember *PrtCnvDbThread::addMember(const int a_identity, const int a_ISSI)
{
PrtCnvMember *newMember = new PrtCnvMember(a_identity, a_ISSI);
{
QMutexLocker lSql(&m_SqlCmdListMutex);
QMutexLocker lId(&m_MemberListByIdMutex);
m_MemberListById.insert(std::make_pair(a_identity, newMember));
m_MemberListByISSI.insert(std::make_pair(a_ISSI, newMember));
}
DEBUG_LOG(m_logger) << "Adding member with eqId " << a_identity
<< " and issi " << a_ISSI
<< ", store in db: " << m_DoStorePermanent;
// add also permanent in DB
if (m_DoStorePermanent)
addSqlCommand(QStringLiteral("INSERT INTO %1 (id, issi) VALUES(%2, %3)")
.arg(memberTableName()).arg(newMember->m_Id).arg(newMember->m_ISSI));
return newMember;
}
// ---------------------------------------------------------------
// removes the member in the 2 maps, sorted by id and ISSI
// ---------------------------------------------------------------
void PrtCnvDbThread::removeMember(PrtCnv::PrtCnvMember *a_member)
{
{
QMutexLocker lSql(&m_SqlCmdListMutex);
QMutexLocker lId(&m_MemberListByIdMutex);
auto iterId = m_MemberListById.find(a_member->m_Id);
if (iterId != m_MemberListById.end())
m_MemberListById.erase(iterId);
else
ERROR_LOG(m_logger) << "Internal error: did not find the member by ID";
auto iterIssi = m_MemberListByISSI.find(a_member->m_ISSI);
if (iterIssi != m_MemberListByISSI.end())
m_MemberListByISSI.erase(iterIssi);
else
ERROR_LOG(m_logger) << "Internal error: did not find the member by ISSI";
}
DEBUG_LOG(m_logger) << "Removing member with eqId " << a_member->m_Id
<< " and issi " << a_member->m_ISSI
<< ", store in db: " << m_DoStorePermanent;
// remove also permanent in DB
if (m_DoStorePermanent)
addSqlCommand(QStringLiteral("DELETE FROM %1 WHERE id = %2 OR issi = %3")
.arg(memberTableName()).arg(a_member->m_Id).arg(a_member->m_ISSI));
delete a_member;
}
// ---------------------------------------------------------------
// Licenses the member
// ---------------------------------------------------------------
bool PrtCnvDbThread::licenseMember(PrtCnvMember *a_member, const bool a_doLicense)
{
if (!a_member)
return false;
a_member->m_Licensed = a_doLicense;
if (m_DoStorePermanent)
addSqlCommand(QStringLiteral("UPDATE %1 SET licensed = %2 WHERE id = %3")
.arg(memberTableName()).arg(a_doLicense ? 1 : 0).arg(a_member->m_Id));
return true;
}
void PrtCnvDbThread::updateExerciseIdForMember(PrtCnvMember *a_member, int newExerciseId)
{
if (!a_member)
return;
a_member->m_loggedInExerciseId = newExerciseId;
DEBUG_LOG(m_logger) << "Updating exerciseId member with eqId " << a_member->m_Id
<< " and issi " << a_member->m_ISSI << " to " << newExerciseId
<< ", store in db: " << m_DoStorePermanent;
if (m_DoStorePermanent)
addSqlCommand(QStringLiteral("UPDATE %1 SET exercise_id = %2 WHERE id = %3")
.arg(memberTableName()).arg(newExerciseId).arg(a_member->m_Id));
}
// ---------------------------------------------------------------
// returns NULL if member not found
// ---------------------------------------------------------------
PrtCnv::PrtCnvMember *PrtCnvDbThread::getMemberById(const uint16_t a_id)
{
QMutexLocker l(&m_MemberListByIdMutex);
auto iter = m_MemberListById.find(a_id);
if (iter != m_MemberListById.end())
return iter->second;
return nullptr; // no member found
}
// ---------------------------------------------------------------
// returns NULL if no member found
// ---------------------------------------------------------------
PrtCnv::PrtCnvMember *PrtCnvDbThread::getMemberByISSI(const uint32_t a_ISSI) const
{
auto iter = m_MemberListByISSI.find(a_ISSI);
if (iter != m_MemberListByISSI.end())
return iter->second;
return nullptr;
}
// ---------------------------------------------------------------
// Returns the actual reset-counter
// ---------------------------------------------------------------
uint8_t PrtCnvDbThread::getResetCounter() const
{
return m_ResetCounter;
}
// ---------------------------------------------------------------
// Sets the actual reset-counter
// ---------------------------------------------------------------
void PrtCnvDbThread::setResetCounter(uint8_t newVal)
{
m_ResetCounter = newVal;
if (m_DoStorePermanent)
{
NSqlQuery q(m_db);
q.exec(QStringLiteral("UPDATE Status SET reset_counter = %1 WHERE id = 1")
.arg(int(newVal)));
}
INFO_LOG(m_logger) << "Update resetCounter=" << (int)m_ResetCounter << " doStorePermanent=" << m_DoStorePermanent;
}
// ---------------------------------------------------------------
// The actual 'Betriebszustände Gefechtsfeld' which will be
// sent to TNEs periodically
// ---------------------------------------------------------------
uint8_t PrtCnvDbThread::getBattleFieldOperatingState() const
{
QMutexLocker l(&m_BattleFieldOpStateMutex);
return m_BattleFieldOperatingState;
}
// ---------------------------------------------------------------
// Sets or unsets pause, depending on a_doSet.
// ---------------------------------------------------------------
void PrtCnvDbThread::setPause(const bool a_doSet)
{
QMutexLocker l(&m_BattleFieldOpStateMutex);
if (a_doSet)
m_BattleFieldOperatingState = m_BattleFieldOperatingState | 0x80; // pause
else
m_BattleFieldOperatingState = m_BattleFieldOperatingState & 0x7F; // no pause
if (m_DoStorePermanent)
addSqlCommand(QStringLiteral("UPDATE %1 SET ctas_operating_state = %2 WHERE id = 1")
.arg(statusTableName()).arg(int(m_BattleFieldOperatingState)));
INFO_LOG(m_logger) << "Update battleFieldOperatingState=" << (int)m_BattleFieldOperatingState << " doStorePermanent=" << m_DoStorePermanent;
}
// ---------------------------------------------------------------
// Set the inhibit time for position telegrams (1 .. 31 Seconds)
// ---------------------------------------------------------------
void PrtCnvDbThread::setPositionSendTime(unsigned int sendTime)
{
QMutexLocker l(&m_BattleFieldOpStateMutex);
if (sendTime > 31)
sendTime = 31;
if (sendTime < 1)
sendTime = 1;
// mode zurücksetzen (Bits 2 - 6)
m_BattleFieldOperatingState &= 0x83;
// und neu setzen
m_BattleFieldOperatingState |= (sendTime << 2);
}
// ---------------------------------------------------------------
// Set the current air load to be sent with the Umgebungsmeldung
// ---------------------------------------------------------------
void PrtCnvDbThread::setAirLoad(PrtCnvRadioThread::E_TrafficLoad airLoad)
{
QMutexLocker l(&m_BattleFieldOpStateMutex);
// mode zurücksetzen
m_BattleFieldOperatingState &= 0xFC;
// und neu setzen
m_BattleFieldOperatingState |= (static_cast<uint8_t>(airLoad) & 0x03);
}
// ---------------------------------------------------------------
// Returns true if the resetCounter itself is different than
// the reset-counter of the environment-telegram
// ---------------------------------------------------------------
bool PrtCnvDbThread::isEnvironmentResetCounterOld() const
{
QMutexLocker l(&m_RadioListMutex);
bool isOld = false;
if (m_EnvironmentTelegram)
isOld = getResetCounter() != m_EnvironmentTelegram->m_RadioTelegram->getPayloadDataUInt8(7);
return isOld;
}
// ---------------------------------------------------------------
// Sets the resetcounter of the environment-telegram to the
// actual reset-counter.
// ---------------------------------------------------------------
void PrtCnvDbThread::setResetCounterOnEnvironment()
{
QMutexLocker l(&m_RadioListMutex);
if (m_EnvironmentTelegram)
m_EnvironmentTelegram->m_RadioTelegram->setPayloadDataWithIndex(7, getResetCounter(), 1);
}
// ---------------------------------------------------------------
// Returns the actual Umgebungsmeldung
// ---------------------------------------------------------------
std::shared_ptr<N_CoPrtCnv::CoPrtCnvRadioTelegram> PrtCnvDbThread::getEnvironmentTelegram()
{
QMutexLocker l(&m_RadioListMutex);
if (m_EnvironmentTelegram)
return std::make_shared<N_CoPrtCnv::CoPrtCnvRadioTelegram>(m_EnvironmentTelegram->m_RadioTelegram->data());
return {};
}
// ---------------------------------------------------------------
// Fills the map with nachforderungsstatistic.
// Returns the number of elements in map.
// ---------------------------------------------------------------
std::map<int32_t, int32_t> PrtCnvDbThread::getClaimStatistics() const
{
std::map<int32_t, int32_t> ret;
QMutexLocker l(&m_MemberListByIdMutex);
for (const auto &p : m_MemberListById)
{
if (p.first != 0)
ret.insert(std::make_pair(p.first, p.second->m_ClaimCount));
}
return ret;
}
// ---------------------------------------------------------------
// Increments the counter for nachforderungen of the TNE with
// id a_identity.
// ---------------------------------------------------------------
void PrtCnvDbThread::incClaimCount(int a_identity)
{
QMutexLocker l(&m_MemberListByIdMutex);
auto iter = m_MemberListById.find(a_identity);
if (iter != m_MemberListById.end())
{
iter->second->m_ClaimCount++;
if (m_DoStorePermanent)
addSqlCommand(QStringLiteral("UPDATE %1 SET claims = claims + 1 WHERE id = %2")
.arg(memberTableName()).arg(iter->second->m_Id));
}
else
ERROR_LOG(m_logger) << "Internal error: did not find TNE with id=" << a_identity;
}
// ---------------------------------------------------------------
// Sets the actual machine-state permanent, used to proceed
// cleaning db after a crash.
// ---------------------------------------------------------------
void PrtCnvDbThread::setMachineState(const int a_newMachineState)
{
NSqlQuery q(m_db);
q.exec(QStringLiteral("UPDATE %1 SET machine_state = %2 WHERE id = 1")
.arg(statusTableName()).arg(a_newMachineState));
}
// ---------------------------------------------------------------
// Returns true if protocol-converter was crashed while
// cleanup DB.
// ---------------------------------------------------------------
bool PrtCnvDbThread::mustProceedCleanup() const
{
return m_DoProceedCleanup;
}
// ---------------------------------------------------------------
// Sends a signal storePermanent if not already such a signal
// is pending
// ---------------------------------------------------------------
void PrtCnvDbThread::sendStorePermanentSignal()
{
if (m_StorePermanentPending.testAndSetOrdered(0, 1))
emit doStorePermanent();
}
void PrtCnvDbThread::closeDb()
{
if (m_db.isOpen())
{
m_db.close();
m_db = QSqlDatabase();
QSqlDatabase::removeDatabase(QStringLiteral("PrtCnv"));
}
}
} // namespace PrtCnv