VariableController.cpp
752 lines
| 31.1 KiB
| text/x-c
|
CppLexer
r235 | #include <Variable/Variable.h> | |||
r539 | #include <Variable/VariableAcquisitionWorker.h> | |||
#include <Variable/VariableCacheStrategy.h> | ||||
|
r111 | #include <Variable/VariableController.h> | ||
|
r113 | #include <Variable/VariableModel.h> | ||
r539 | #include <Variable/VariableSynchronizationGroup.h> | |||
|
r111 | |||
|
r166 | #include <Data/DataProviderParameters.h> | ||
#include <Data/IDataProvider.h> | ||||
#include <Data/IDataSeries.h> | ||||
r626 | #include <Data/VariableRequest.h> | |||
r193 | #include <Time/TimeController.h> | |||
|
r166 | |||
|
r111 | #include <QMutex> | ||
#include <QThread> | ||||
|
r376 | #include <QUuid> | ||
r304 | #include <QtCore/QItemSelectionModel> | |||
|
r111 | |||
r626 | #include <deque> | |||
r539 | #include <set> | |||
r225 | #include <unordered_map> | |||
|
r111 | Q_LOGGING_CATEGORY(LOG_VariableController, "VariableController") | ||
r539 | namespace { | |||
r545 | SqpRange computeSynchroRangeRequested(const SqpRange &varRange, const SqpRange &graphRange, | |||
r539 | const SqpRange &oldGraphRange) | |||
{ | ||||
r545 | auto zoomType = VariableController::getZoomType(graphRange, oldGraphRange); | |||
r539 | ||||
auto varRangeRequested = varRange; | ||||
switch (zoomType) { | ||||
case AcquisitionZoomType::ZoomIn: { | ||||
r545 | auto deltaLeft = graphRange.m_TStart - oldGraphRange.m_TStart; | |||
auto deltaRight = oldGraphRange.m_TEnd - graphRange.m_TEnd; | ||||
r539 | varRangeRequested.m_TStart += deltaLeft; | |||
varRangeRequested.m_TEnd -= deltaRight; | ||||
break; | ||||
} | ||||
case AcquisitionZoomType::ZoomOut: { | ||||
r545 | auto deltaLeft = oldGraphRange.m_TStart - graphRange.m_TStart; | |||
auto deltaRight = graphRange.m_TEnd - oldGraphRange.m_TEnd; | ||||
r539 | varRangeRequested.m_TStart -= deltaLeft; | |||
varRangeRequested.m_TEnd += deltaRight; | ||||
break; | ||||
} | ||||
case AcquisitionZoomType::PanRight: { | ||||
r545 | auto deltaRight = graphRange.m_TEnd - oldGraphRange.m_TEnd; | |||
r539 | varRangeRequested.m_TStart += deltaRight; | |||
varRangeRequested.m_TEnd += deltaRight; | ||||
break; | ||||
} | ||||
case AcquisitionZoomType::PanLeft: { | ||||
r545 | auto deltaLeft = oldGraphRange.m_TStart - graphRange.m_TStart; | |||
r539 | varRangeRequested.m_TStart -= deltaLeft; | |||
varRangeRequested.m_TEnd -= deltaLeft; | ||||
break; | ||||
} | ||||
case AcquisitionZoomType::Unknown: { | ||||
qCCritical(LOG_VariableController()) | ||||
<< VariableController::tr("Impossible to synchronize: zoom type unknown"); | ||||
break; | ||||
} | ||||
default: | ||||
qCCritical(LOG_VariableController()) << VariableController::tr( | ||||
"Impossible to synchronize: zoom type not take into account"); | ||||
// No action | ||||
break; | ||||
} | ||||
return varRangeRequested; | ||||
} | ||||
} | ||||
|
r111 | struct VariableController::VariableControllerPrivate { | ||
|
r159 | explicit VariableControllerPrivate(VariableController *parent) | ||
r225 | : m_WorkingMutex{}, | |||
m_VariableModel{new VariableModel{parent}}, | ||||
r304 | m_VariableSelectionModel{new QItemSelectionModel{m_VariableModel, parent}}, | |||
r539 | m_VariableCacheStrategy{std::make_unique<VariableCacheStrategy>()}, | |||
r571 | m_VariableAcquisitionWorker{std::make_unique<VariableAcquisitionWorker>()}, | |||
q{parent} | ||||
r539 | { | |||
m_VariableAcquisitionWorker->moveToThread(&m_VariableAcquisitionWorkerThread); | ||||
m_VariableAcquisitionWorkerThread.setObjectName("VariableAcquisitionWorkerThread"); | ||||
} | ||||
virtual ~VariableControllerPrivate() | ||||
|
r111 | { | ||
r539 | qCDebug(LOG_VariableController()) << tr("VariableControllerPrivate destruction"); | |||
m_VariableAcquisitionWorkerThread.quit(); | ||||
m_VariableAcquisitionWorkerThread.wait(); | ||||
|
r111 | } | ||
r539 | ||||
r626 | void processRequest(std::shared_ptr<Variable> var, const SqpRange &rangeRequested, | |||
QUuid varRequestId); | ||||
r539 | ||||
r530 | QVector<SqpRange> provideNotInCacheDateTimeList(std::shared_ptr<Variable> variable, | |||
const SqpRange &dateTime); | ||||
r529 | ||||
r539 | std::shared_ptr<Variable> findVariable(QUuid vIdentifier); | |||
std::shared_ptr<IDataSeries> | ||||
retrieveDataSeries(const QVector<AcquisitionDataPacket> acqDataPacketVector); | ||||
void registerProvider(std::shared_ptr<IDataProvider> provider); | ||||
r626 | void storeVariableRequest(QUuid varId, QUuid varRequestId, const VariableRequest &varRequest); | |||
QUuid acceptVariableRequest(QUuid varId, std::shared_ptr<IDataSeries> dataSeries); | ||||
void updateVariableRequest(QUuid varRequestId); | ||||
void cancelVariableRequest(QUuid varRequestId); | ||||
|
r111 | QMutex m_WorkingMutex; | ||
|
r159 | /// Variable model. The VariableController has the ownership | ||
VariableModel *m_VariableModel; | ||||
r304 | QItemSelectionModel *m_VariableSelectionModel; | |||
r193 | ||||
r225 | ||||
r195 | TimeController *m_TimeController{nullptr}; | |||
r539 | std::unique_ptr<VariableCacheStrategy> m_VariableCacheStrategy; | |||
std::unique_ptr<VariableAcquisitionWorker> m_VariableAcquisitionWorker; | ||||
QThread m_VariableAcquisitionWorkerThread; | ||||
r235 | ||||
std::unordered_map<std::shared_ptr<Variable>, std::shared_ptr<IDataProvider> > | ||||
m_VariableToProviderMap; | ||||
|
r420 | std::unordered_map<std::shared_ptr<Variable>, QUuid> m_VariableToIdentifierMap; | ||
r539 | std::map<QUuid, std::shared_ptr<VariableSynchronizationGroup> > | |||
m_GroupIdToVariableSynchronizationGroupMap; | ||||
std::map<QUuid, QUuid> m_VariableIdGroupIdMap; | ||||
std::set<std::shared_ptr<IDataProvider> > m_ProviderSet; | ||||
r571 | ||||
r626 | std::map<QUuid, std::map<QUuid, VariableRequest> > m_VarRequestIdToVarIdVarRequestMap; | |||
std::map<QUuid, std::deque<QUuid> > m_VarIdToVarRequestIdQueueMap; | ||||
r571 | ||||
VariableController *q; | ||||
|
r111 | }; | ||
r539 | ||||
|
r111 | VariableController::VariableController(QObject *parent) | ||
|
r159 | : QObject{parent}, impl{spimpl::make_unique_impl<VariableControllerPrivate>(this)} | ||
|
r111 | { | ||
r365 | qCDebug(LOG_VariableController()) << tr("VariableController construction") | |||
<< QThread::currentThread(); | ||||
r422 | ||||
connect(impl->m_VariableModel, &VariableModel::abortProgessRequested, this, | ||||
&VariableController::onAbortProgressRequested); | ||||
r539 | ||||
connect(impl->m_VariableAcquisitionWorker.get(), &VariableAcquisitionWorker::dataProvided, this, | ||||
&VariableController::onDataProvided); | ||||
connect(impl->m_VariableAcquisitionWorker.get(), | ||||
&VariableAcquisitionWorker::variableRequestInProgress, this, | ||||
&VariableController::onVariableRetrieveDataInProgress); | ||||
connect(&impl->m_VariableAcquisitionWorkerThread, &QThread::started, | ||||
impl->m_VariableAcquisitionWorker.get(), &VariableAcquisitionWorker::initialize); | ||||
connect(&impl->m_VariableAcquisitionWorkerThread, &QThread::finished, | ||||
impl->m_VariableAcquisitionWorker.get(), &VariableAcquisitionWorker::finalize); | ||||
impl->m_VariableAcquisitionWorkerThread.start(); | ||||
|
r111 | } | ||
VariableController::~VariableController() | ||||
{ | ||||
r365 | qCDebug(LOG_VariableController()) << tr("VariableController destruction") | |||
<< QThread::currentThread(); | ||||
|
r111 | this->waitForFinish(); | ||
} | ||||
|
r166 | VariableModel *VariableController::variableModel() noexcept | ||
|
r113 | { | ||
|
r166 | return impl->m_VariableModel; | ||
|
r113 | } | ||
r304 | QItemSelectionModel *VariableController::variableSelectionModel() noexcept | |||
{ | ||||
return impl->m_VariableSelectionModel; | ||||
} | ||||
r193 | void VariableController::setTimeController(TimeController *timeController) noexcept | |||
{ | ||||
impl->m_TimeController = timeController; | ||||
} | ||||
|
r706 | std::shared_ptr<Variable> | ||
VariableController::cloneVariable(std::shared_ptr<Variable> variable) noexcept | ||||
{ | ||||
|
r707 | // Clones variable | ||
auto duplicate = variable->clone(); | ||||
return duplicate; | ||||
|
r706 | } | ||
|
r329 | void VariableController::deleteVariable(std::shared_ptr<Variable> variable) noexcept | ||
{ | ||||
if (!variable) { | ||||
qCCritical(LOG_VariableController()) << "Can't delete variable: variable is null"; | ||||
return; | ||||
} | ||||
|
r335 | // Spreads in SciQlop that the variable will be deleted, so that potential receivers can | ||
// make some treatments before the deletion | ||||
emit variableAboutToBeDeleted(variable); | ||||
|
r330 | |||
|
r420 | // Deletes identifier | ||
impl->m_VariableToIdentifierMap.erase(variable); | ||||
|
r330 | // Deletes provider | ||
auto nbProvidersDeleted = impl->m_VariableToProviderMap.erase(variable); | ||||
qCDebug(LOG_VariableController()) | ||||
<< tr("Number of providers deleted for variable %1: %2") | ||||
.arg(variable->name(), QString::number(nbProvidersDeleted)); | ||||
|
r329 | |||
|
r331 | |||
|
r332 | // Deletes from model | ||
impl->m_VariableModel->deleteVariable(variable); | ||||
} | ||||
|
r329 | |||
void VariableController::deleteVariables( | ||||
const QVector<std::shared_ptr<Variable> > &variables) noexcept | ||||
{ | ||||
for (auto variable : qAsConst(variables)) { | ||||
deleteVariable(variable); | ||||
} | ||||
} | ||||
r422 | void VariableController::abortProgress(std::shared_ptr<Variable> variable) | |||
{ | ||||
} | ||||
r590 | std::shared_ptr<Variable> | |||
VariableController::createVariable(const QString &name, const QVariantHash &metadata, | ||||
std::shared_ptr<IDataProvider> provider) noexcept | ||||
|
r152 | { | ||
r193 | if (!impl->m_TimeController) { | |||
qCCritical(LOG_VariableController()) | ||||
<< tr("Impossible to create variable: The time controller is null"); | ||||
r590 | return nullptr; | |||
r193 | } | |||
r539 | auto range = impl->m_TimeController->dateTime(); | |||
|
r410 | |||
r539 | if (auto newVariable = impl->m_VariableModel->createVariable(name, range, metadata)) { | |||
r389 | auto identifier = QUuid::createUuid(); | |||
r225 | ||||
r235 | // store the provider | |||
r539 | impl->registerProvider(provider); | |||
// Associate the provider | ||||
r235 | impl->m_VariableToProviderMap[newVariable] = provider; | |||
|
r420 | impl->m_VariableToIdentifierMap[newVariable] = identifier; | ||
r318 | ||||
r626 | auto varRequestId = QUuid::createUuid(); | |||
qCInfo(LOG_VariableController()) << "processRequest for" << name << varRequestId; | ||||
impl->processRequest(newVariable, range, varRequestId); | ||||
impl->updateVariableRequest(varRequestId); | ||||
r590 | ||||
return newVariable; | ||||
|
r166 | } | ||
|
r152 | } | ||
r512 | void VariableController::onDateTimeOnSelection(const SqpRange &dateTime) | |||
r304 | { | |||
r626 | // TODO check synchronisation and Rescale | |||
r365 | qCDebug(LOG_VariableController()) << "VariableController::onDateTimeOnSelection" | |||
<< QThread::currentThread()->objectName(); | ||||
r304 | auto selectedRows = impl->m_VariableSelectionModel->selectedRows(); | |||
r626 | auto varRequestId = QUuid::createUuid(); | |||
r304 | ||||
for (const auto &selectedRow : qAsConst(selectedRows)) { | ||||
if (auto selectedVariable = impl->m_VariableModel->variable(selectedRow.row())) { | ||||
r539 | selectedVariable->setRange(dateTime); | |||
r626 | impl->processRequest(selectedVariable, dateTime, varRequestId); | |||
r442 | ||||
// notify that rescale operation has to be done | ||||
r437 | emit rangeChanged(selectedVariable, dateTime); | |||
r304 | } | |||
} | ||||
r626 | impl->updateVariableRequest(varRequestId); | |||
r304 | } | |||
r539 | void VariableController::onDataProvided(QUuid vIdentifier, const SqpRange &rangeRequested, | |||
const SqpRange &cacheRangeRequested, | ||||
QVector<AcquisitionDataPacket> dataAcquired) | ||||
r401 | { | |||
r626 | auto retrievedDataSeries = impl->retrieveDataSeries(dataAcquired); | |||
auto varRequestId = impl->acceptVariableRequest(vIdentifier, retrievedDataSeries); | ||||
if (!varRequestId.isNull()) { | ||||
impl->updateVariableRequest(varRequestId); | ||||
r539 | } | |||
} | ||||
r401 | ||||
r539 | void VariableController::onVariableRetrieveDataInProgress(QUuid identifier, double progress) | |||
{ | ||||
r546 | if (auto var = impl->findVariable(identifier)) { | |||
r539 | impl->m_VariableModel->setDataProgress(var, progress); | |||
} | ||||
else { | ||||
qCCritical(LOG_VariableController()) | ||||
<< tr("Impossible to notify progression of a null variable"); | ||||
r401 | } | |||
} | ||||
r422 | void VariableController::onAbortProgressRequested(std::shared_ptr<Variable> variable) | |||
{ | ||||
qCDebug(LOG_VariableController()) << "TORM: VariableController::onAbortProgressRequested" | ||||
<< QThread::currentThread()->objectName(); | ||||
r423 | auto it = impl->m_VariableToIdentifierMap.find(variable); | |||
if (it != impl->m_VariableToIdentifierMap.cend()) { | ||||
r422 | impl->m_VariableToProviderMap.at(variable)->requestDataAborting(it->second); | |||
} | ||||
else { | ||||
qCWarning(LOG_VariableController()) | ||||
<< tr("Aborting progression of inexistant variable detected !!!") | ||||
<< QThread::currentThread()->objectName(); | ||||
} | ||||
} | ||||
r539 | void VariableController::onAddSynchronizationGroupId(QUuid synchronizationGroupId) | |||
{ | ||||
r540 | qCDebug(LOG_VariableController()) << "TORM: VariableController::onAddSynchronizationGroupId" | |||
<< QThread::currentThread()->objectName() | ||||
<< synchronizationGroupId; | ||||
r539 | auto vSynchroGroup = std::make_shared<VariableSynchronizationGroup>(); | |||
impl->m_GroupIdToVariableSynchronizationGroupMap.insert( | ||||
std::make_pair(synchronizationGroupId, vSynchroGroup)); | ||||
} | ||||
void VariableController::onRemoveSynchronizationGroupId(QUuid synchronizationGroupId) | ||||
{ | ||||
impl->m_GroupIdToVariableSynchronizationGroupMap.erase(synchronizationGroupId); | ||||
} | ||||
r540 | void VariableController::onAddSynchronized(std::shared_ptr<Variable> variable, | |||
QUuid synchronizationGroupId) | ||||
{ | ||||
qCDebug(LOG_VariableController()) << "TORM: VariableController::onAddSynchronized" | ||||
<< synchronizationGroupId; | ||||
r546 | auto varToVarIdIt = impl->m_VariableToIdentifierMap.find(variable); | |||
if (varToVarIdIt != impl->m_VariableToIdentifierMap.cend()) { | ||||
auto groupIdToVSGIt | ||||
r540 | = impl->m_GroupIdToVariableSynchronizationGroupMap.find(synchronizationGroupId); | |||
r546 | if (groupIdToVSGIt != impl->m_GroupIdToVariableSynchronizationGroupMap.cend()) { | |||
r540 | impl->m_VariableIdGroupIdMap.insert( | |||
r546 | std::make_pair(varToVarIdIt->second, synchronizationGroupId)); | |||
groupIdToVSGIt->second->addVariableId(varToVarIdIt->second); | ||||
r540 | } | |||
else { | ||||
qCCritical(LOG_VariableController()) | ||||
<< tr("Impossible to synchronize a variable with an unknown sycnhronization group") | ||||
<< variable->name(); | ||||
} | ||||
} | ||||
else { | ||||
qCCritical(LOG_VariableController()) | ||||
<< tr("Impossible to synchronize a variable with no identifier") << variable->name(); | ||||
} | ||||
} | ||||
r235 | ||||
r539 | void VariableController::onRequestDataLoading(QVector<std::shared_ptr<Variable> > variables, | |||
const SqpRange &range, const SqpRange &oldRange, | ||||
bool synchronise) | ||||
r235 | { | |||
r539 | // NOTE: oldRange isn't really necessary since oldRange == variable->range(). | |||
r235 | // we want to load data of the variable for the dateTime. | |||
// First we check if the cache contains some of them. | ||||
// For the other, we ask the provider to give them. | ||||
r626 | auto varRequestId = QUuid::createUuid(); | |||
r627 | qCInfo(LOG_VariableController()) << "VariableController::onRequestDataLoading" | |||
<< QThread::currentThread()->objectName() << varRequestId; | ||||
r626 | ||||
r546 | for (const auto &var : variables) { | |||
r627 | qCDebug(LOG_VariableController()) << "processRequest for" << var->name() << varRequestId; | |||
r626 | impl->processRequest(var, range, varRequestId); | |||
r539 | } | |||
r235 | ||||
r539 | if (synchronise) { | |||
// Get the group ids | ||||
r540 | qCDebug(LOG_VariableController()) | |||
r546 | << "TORM VariableController::onRequestDataLoading for synchro var ENABLE"; | |||
r627 | auto groupIds = std::set<QUuid>{}; | |||
auto groupIdToOldRangeMap = std::map<QUuid, SqpRange>{}; | ||||
r546 | for (const auto &var : variables) { | |||
r540 | auto varToVarIdIt = impl->m_VariableToIdentifierMap.find(var); | |||
if (varToVarIdIt != impl->m_VariableToIdentifierMap.cend()) { | ||||
auto vId = varToVarIdIt->second; | ||||
auto varIdToGroupIdIt = impl->m_VariableIdGroupIdMap.find(vId); | ||||
if (varIdToGroupIdIt != impl->m_VariableIdGroupIdMap.cend()) { | ||||
auto gId = varIdToGroupIdIt->second; | ||||
r627 | groupIdToOldRangeMap.insert(std::make_pair(gId, var->range())); | |||
r539 | if (groupIds.find(gId) == groupIds.cend()) { | |||
r540 | qCDebug(LOG_VariableController()) << "Synchro detect group " << gId; | |||
r539 | groupIds.insert(gId); | |||
} | ||||
} | ||||
} | ||||
r298 | } | |||
r539 | ||||
// We assume here all group ids exist | ||||
r546 | for (const auto &gId : groupIds) { | |||
r539 | auto vSynchronizationGroup = impl->m_GroupIdToVariableSynchronizationGroupMap.at(gId); | |||
auto vSyncIds = vSynchronizationGroup->getIds(); | ||||
r540 | qCDebug(LOG_VariableController()) << "Var in synchro group "; | |||
r539 | for (auto vId : vSyncIds) { | |||
auto var = impl->findVariable(vId); | ||||
r540 | ||||
// Don't process already processed var | ||||
if (!variables.contains(var)) { | ||||
if (var != nullptr) { | ||||
qCDebug(LOG_VariableController()) << "processRequest synchro for" | ||||
<< var->name(); | ||||
r627 | auto vSyncRangeRequested = computeSynchroRangeRequested( | |||
var->range(), range, groupIdToOldRangeMap.at(gId)); | ||||
qCDebug(LOG_VariableController()) << "synchro RR" << vSyncRangeRequested; | ||||
r626 | impl->processRequest(var, vSyncRangeRequested, varRequestId); | |||
r540 | } | |||
else { | ||||
qCCritical(LOG_VariableController()) | ||||
<< tr("Impossible to synchronize a null variable"); | ||||
} | ||||
r539 | } | |||
} | ||||
r298 | } | |||
r235 | } | |||
r626 | ||||
impl->updateVariableRequest(varRequestId); | ||||
r235 | } | |||
|
r111 | void VariableController::initialize() | ||
{ | ||||
qCDebug(LOG_VariableController()) << tr("VariableController init") << QThread::currentThread(); | ||||
impl->m_WorkingMutex.lock(); | ||||
qCDebug(LOG_VariableController()) << tr("VariableController init END"); | ||||
} | ||||
void VariableController::finalize() | ||||
{ | ||||
impl->m_WorkingMutex.unlock(); | ||||
} | ||||
void VariableController::waitForFinish() | ||||
{ | ||||
QMutexLocker locker{&impl->m_WorkingMutex}; | ||||
} | ||||
r529 | ||||
r539 | AcquisitionZoomType VariableController::getZoomType(const SqpRange &range, const SqpRange &oldRange) | |||
{ | ||||
// t1.m_TStart <= t2.m_TStart && t2.m_TEnd <= t1.m_TEnd | ||||
auto zoomType = AcquisitionZoomType::Unknown; | ||||
if (range.m_TStart <= oldRange.m_TStart && oldRange.m_TEnd <= range.m_TEnd) { | ||||
zoomType = AcquisitionZoomType::ZoomOut; | ||||
} | ||||
else if (range.m_TStart > oldRange.m_TStart && range.m_TEnd > oldRange.m_TEnd) { | ||||
zoomType = AcquisitionZoomType::PanRight; | ||||
} | ||||
else if (range.m_TStart < oldRange.m_TStart && range.m_TEnd < oldRange.m_TEnd) { | ||||
zoomType = AcquisitionZoomType::PanLeft; | ||||
} | ||||
else if (range.m_TStart > oldRange.m_TStart && oldRange.m_TEnd > range.m_TEnd) { | ||||
zoomType = AcquisitionZoomType::ZoomIn; | ||||
} | ||||
else { | ||||
qCCritical(LOG_VariableController()) << "getZoomType: Unknown type detected"; | ||||
} | ||||
return zoomType; | ||||
} | ||||
r529 | ||||
r539 | void VariableController::VariableControllerPrivate::processRequest(std::shared_ptr<Variable> var, | |||
r626 | const SqpRange &rangeRequested, | |||
QUuid varRequestId) | ||||
r529 | { | |||
r626 | // TODO: protect at | |||
auto varRequest = VariableRequest{}; | ||||
auto varId = m_VariableToIdentifierMap.at(var); | ||||
auto varStrategyRangesRequested | ||||
= m_VariableCacheStrategy->computeStrategyRanges(var->range(), rangeRequested); | ||||
auto notInCacheRangeList = var->provideNotInCacheRangeList(varStrategyRangesRequested.second); | ||||
auto inCacheRangeList = var->provideInCacheRangeList(varStrategyRangesRequested.second); | ||||
r539 | ||||
if (!notInCacheRangeList.empty()) { | ||||
r626 | varRequest.m_RangeRequested = varStrategyRangesRequested.first; | |||
varRequest.m_CacheRangeRequested = varStrategyRangesRequested.second; | ||||
r627 | qCDebug(LOG_VariableAcquisitionWorker()) << tr("TORM processRequest RR ") << rangeRequested; | |||
qCDebug(LOG_VariableAcquisitionWorker()) << tr("TORM processRequest R ") | ||||
<< varStrategyRangesRequested.first; | ||||
qCDebug(LOG_VariableAcquisitionWorker()) << tr("TORM processRequest CR ") | ||||
<< varStrategyRangesRequested.second; | ||||
r626 | // store VarRequest | |||
storeVariableRequest(varId, varRequestId, varRequest); | ||||
r539 | auto varProvider = m_VariableToProviderMap.at(var); | |||
if (varProvider != nullptr) { | ||||
r626 | auto varRequestIdCanceled = m_VariableAcquisitionWorker->pushVariableRequest( | |||
varRequestId, varId, varStrategyRangesRequested.first, | ||||
varStrategyRangesRequested.second, | ||||
r539 | DataProviderParameters{std::move(notInCacheRangeList), var->metadata()}, | |||
varProvider); | ||||
r626 | ||||
if (!varRequestIdCanceled.isNull()) { | ||||
r627 | qCInfo(LOG_VariableAcquisitionWorker()) << tr("varRequestIdCanceled: ") | |||
<< varRequestIdCanceled; | ||||
r626 | cancelVariableRequest(varRequestIdCanceled); | |||
} | ||||
r529 | } | |||
else { | ||||
r539 | qCCritical(LOG_VariableController()) | |||
<< "Impossible to provide data with a null provider"; | ||||
r529 | } | |||
r571 | ||||
if (!inCacheRangeList.empty()) { | ||||
emit q->updateVarDisplaying(var, inCacheRangeList.first()); | ||||
} | ||||
r529 | } | |||
r539 | else { | |||
r626 | ||||
varRequest.m_RangeRequested = varStrategyRangesRequested.first; | ||||
varRequest.m_CacheRangeRequested = varStrategyRangesRequested.second; | ||||
// store VarRequest | ||||
storeVariableRequest(varId, varRequestId, varRequest); | ||||
acceptVariableRequest(varId, | ||||
var->dataSeries()->subDataSeries(varStrategyRangesRequested.second)); | ||||
r539 | } | |||
} | ||||
std::shared_ptr<Variable> | ||||
VariableController::VariableControllerPrivate::findVariable(QUuid vIdentifier) | ||||
{ | ||||
std::shared_ptr<Variable> var; | ||||
auto findReply = [vIdentifier](const auto &entry) { return vIdentifier == entry.second; }; | ||||
auto end = m_VariableToIdentifierMap.cend(); | ||||
auto it = std::find_if(m_VariableToIdentifierMap.cbegin(), end, findReply); | ||||
if (it != end) { | ||||
var = it->first; | ||||
} | ||||
else { | ||||
qCCritical(LOG_VariableController()) | ||||
<< tr("Impossible to find the variable with the identifier: ") << vIdentifier; | ||||
} | ||||
r529 | ||||
r539 | return var; | |||
} | ||||
std::shared_ptr<IDataSeries> VariableController::VariableControllerPrivate::retrieveDataSeries( | ||||
const QVector<AcquisitionDataPacket> acqDataPacketVector) | ||||
{ | ||||
r542 | qCDebug(LOG_VariableController()) << tr("TORM: retrieveDataSeries acqDataPacketVector size") | |||
<< acqDataPacketVector.size(); | ||||
r539 | std::shared_ptr<IDataSeries> dataSeries; | |||
if (!acqDataPacketVector.isEmpty()) { | ||||
dataSeries = acqDataPacketVector[0].m_DateSeries; | ||||
for (int i = 1; i < acqDataPacketVector.size(); ++i) { | ||||
dataSeries->merge(acqDataPacketVector[i].m_DateSeries.get()); | ||||
} | ||||
} | ||||
r542 | qCDebug(LOG_VariableController()) << tr("TORM: retrieveDataSeries acqDataPacketVector size END") | |||
<< acqDataPacketVector.size(); | ||||
r539 | return dataSeries; | |||
} | ||||
void VariableController::VariableControllerPrivate::registerProvider( | ||||
std::shared_ptr<IDataProvider> provider) | ||||
{ | ||||
if (m_ProviderSet.find(provider) == m_ProviderSet.end()) { | ||||
r542 | qCDebug(LOG_VariableController()) << tr("Registering of a new provider") | |||
<< provider->objectName(); | ||||
r539 | m_ProviderSet.insert(provider); | |||
connect(provider.get(), &IDataProvider::dataProvided, m_VariableAcquisitionWorker.get(), | ||||
&VariableAcquisitionWorker::onVariableDataAcquired); | ||||
connect(provider.get(), &IDataProvider::dataProvidedProgress, | ||||
m_VariableAcquisitionWorker.get(), | ||||
&VariableAcquisitionWorker::onVariableRetrieveDataInProgress); | ||||
} | ||||
else { | ||||
r540 | qCDebug(LOG_VariableController()) << tr("Cannot register provider, it already exists "); | |||
r539 | } | |||
r529 | } | |||
r626 | ||||
void VariableController::VariableControllerPrivate::storeVariableRequest( | ||||
QUuid varId, QUuid varRequestId, const VariableRequest &varRequest) | ||||
{ | ||||
// First request for the variable. we can create an entry for it | ||||
auto varIdToVarRequestIdQueueMapIt = m_VarIdToVarRequestIdQueueMap.find(varId); | ||||
if (varIdToVarRequestIdQueueMapIt == m_VarIdToVarRequestIdQueueMap.cend()) { | ||||
auto varRequestIdQueue = std::deque<QUuid>{}; | ||||
qCDebug(LOG_VariableController()) << tr("Store REQUEST in QUEUE"); | ||||
varRequestIdQueue.push_back(varRequestId); | ||||
m_VarIdToVarRequestIdQueueMap.insert(std::make_pair(varId, std::move(varRequestIdQueue))); | ||||
} | ||||
else { | ||||
qCDebug(LOG_VariableController()) << tr("Store REQUEST in EXISTING QUEUE"); | ||||
auto &varRequestIdQueue = varIdToVarRequestIdQueueMapIt->second; | ||||
varRequestIdQueue.push_back(varRequestId); | ||||
} | ||||
auto varRequestIdToVarIdVarRequestMapIt = m_VarRequestIdToVarIdVarRequestMap.find(varRequestId); | ||||
if (varRequestIdToVarIdVarRequestMapIt == m_VarRequestIdToVarIdVarRequestMap.cend()) { | ||||
auto varIdToVarRequestMap = std::map<QUuid, VariableRequest>{}; | ||||
varIdToVarRequestMap.insert(std::make_pair(varId, varRequest)); | ||||
qCDebug(LOG_VariableController()) << tr("Store REQUESTID in MAP"); | ||||
m_VarRequestIdToVarIdVarRequestMap.insert( | ||||
std::make_pair(varRequestId, std::move(varIdToVarRequestMap))); | ||||
} | ||||
else { | ||||
auto &varIdToVarRequestMap = varRequestIdToVarIdVarRequestMapIt->second; | ||||
qCDebug(LOG_VariableController()) << tr("Store REQUESTID in EXISTING MAP"); | ||||
varIdToVarRequestMap.insert(std::make_pair(varId, varRequest)); | ||||
} | ||||
} | ||||
QUuid VariableController::VariableControllerPrivate::acceptVariableRequest( | ||||
QUuid varId, std::shared_ptr<IDataSeries> dataSeries) | ||||
{ | ||||
QUuid varRequestId; | ||||
auto varIdToVarRequestIdQueueMapIt = m_VarIdToVarRequestIdQueueMap.find(varId); | ||||
if (varIdToVarRequestIdQueueMapIt != m_VarIdToVarRequestIdQueueMap.cend()) { | ||||
auto &varRequestIdQueue = varIdToVarRequestIdQueueMapIt->second; | ||||
varRequestId = varRequestIdQueue.front(); | ||||
auto varRequestIdToVarIdVarRequestMapIt | ||||
= m_VarRequestIdToVarIdVarRequestMap.find(varRequestId); | ||||
if (varRequestIdToVarIdVarRequestMapIt != m_VarRequestIdToVarIdVarRequestMap.cend()) { | ||||
auto &varIdToVarRequestMap = varRequestIdToVarIdVarRequestMapIt->second; | ||||
auto varIdToVarRequestMapIt = varIdToVarRequestMap.find(varId); | ||||
if (varIdToVarRequestMapIt != varIdToVarRequestMap.cend()) { | ||||
qCDebug(LOG_VariableController()) << tr("acceptVariableRequest"); | ||||
auto &varRequest = varIdToVarRequestMapIt->second; | ||||
varRequest.m_DataSeries = dataSeries; | ||||
varRequest.m_CanUpdate = true; | ||||
} | ||||
else { | ||||
qCDebug(LOG_VariableController()) | ||||
<< tr("Impossible to acceptVariableRequest of a unknown variable id attached " | ||||
"to a variableRequestId") | ||||
<< varRequestId << varId; | ||||
} | ||||
} | ||||
else { | ||||
qCCritical(LOG_VariableController()) | ||||
<< tr("Impossible to acceptVariableRequest of a unknown variableRequestId") | ||||
<< varRequestId; | ||||
} | ||||
qCDebug(LOG_VariableController()) << tr("1: erase REQUEST in QUEUE ?") | ||||
<< varRequestIdQueue.size(); | ||||
varRequestIdQueue.pop_front(); | ||||
qCDebug(LOG_VariableController()) << tr("2: erase REQUEST in QUEUE ?") | ||||
<< varRequestIdQueue.size(); | ||||
if (varRequestIdQueue.empty()) { | ||||
m_VarIdToVarRequestIdQueueMap.erase(varId); | ||||
} | ||||
} | ||||
else { | ||||
qCCritical(LOG_VariableController()) | ||||
<< tr("Impossible to acceptVariableRequest of a unknown variable id") << varId; | ||||
} | ||||
return varRequestId; | ||||
} | ||||
void VariableController::VariableControllerPrivate::updateVariableRequest(QUuid varRequestId) | ||||
{ | ||||
auto varRequestIdToVarIdVarRequestMapIt = m_VarRequestIdToVarIdVarRequestMap.find(varRequestId); | ||||
if (varRequestIdToVarIdVarRequestMapIt != m_VarRequestIdToVarIdVarRequestMap.cend()) { | ||||
bool processVariableUpdate = true; | ||||
auto &varIdToVarRequestMap = varRequestIdToVarIdVarRequestMapIt->second; | ||||
for (auto varIdToVarRequestMapIt = varIdToVarRequestMap.cbegin(); | ||||
(varIdToVarRequestMapIt != varIdToVarRequestMap.cend()) && processVariableUpdate; | ||||
++varIdToVarRequestMapIt) { | ||||
processVariableUpdate &= varIdToVarRequestMapIt->second.m_CanUpdate; | ||||
qCDebug(LOG_VariableController()) << tr("updateVariableRequest") | ||||
<< processVariableUpdate; | ||||
} | ||||
if (processVariableUpdate) { | ||||
for (auto varIdToVarRequestMapIt = varIdToVarRequestMap.cbegin(); | ||||
varIdToVarRequestMapIt != varIdToVarRequestMap.cend(); ++varIdToVarRequestMapIt) { | ||||
if (auto var = findVariable(varIdToVarRequestMapIt->first)) { | ||||
auto &varRequest = varIdToVarRequestMapIt->second; | ||||
var->setRange(varRequest.m_RangeRequested); | ||||
var->setCacheRange(varRequest.m_CacheRangeRequested); | ||||
r627 | qCDebug(LOG_VariableController()) << tr("1: onDataProvided") | |||
<< varRequest.m_RangeRequested; | ||||
qCDebug(LOG_VariableController()) << tr("2: onDataProvided") | ||||
<< varRequest.m_CacheRangeRequested; | ||||
r626 | var->mergeDataSeries(varRequest.m_DataSeries); | |||
r627 | qCDebug(LOG_VariableController()) << tr("3: onDataProvided") | |||
<< varRequest.m_DataSeries->range(); | ||||
r626 | qCDebug(LOG_VariableController()) << tr("4: onDataProvided"); | |||
|
r702 | |||
/// @todo MPL: confirm | ||||
// Variable update is notified only if there is no pending request for it | ||||
if (m_VarIdToVarRequestIdQueueMap.count(varIdToVarRequestMapIt->first) == 0) { | ||||
emit var->updated(); | ||||
} | ||||
r626 | } | |||
else { | ||||
qCCritical(LOG_VariableController()) | ||||
<< tr("Impossible to update data to a null variable"); | ||||
} | ||||
} | ||||
// cleaning varRequestId | ||||
qCDebug(LOG_VariableController()) << tr("0: erase REQUEST in MAP ?") | ||||
<< m_VarRequestIdToVarIdVarRequestMap.size(); | ||||
m_VarRequestIdToVarIdVarRequestMap.erase(varRequestId); | ||||
qCDebug(LOG_VariableController()) << tr("1: erase REQUEST in MAP ?") | ||||
<< m_VarRequestIdToVarIdVarRequestMap.size(); | ||||
} | ||||
} | ||||
else { | ||||
qCCritical(LOG_VariableController()) | ||||
<< tr("Cannot updateVariableRequest for a unknow varRequestId") << varRequestId; | ||||
} | ||||
} | ||||
void VariableController::VariableControllerPrivate::cancelVariableRequest(QUuid varRequestId) | ||||
{ | ||||
// cleaning varRequestId | ||||
m_VarRequestIdToVarIdVarRequestMap.erase(varRequestId); | ||||
for (auto varIdToVarRequestIdQueueMapIt = m_VarIdToVarRequestIdQueueMap.begin(); | ||||
varIdToVarRequestIdQueueMapIt != m_VarIdToVarRequestIdQueueMap.end();) { | ||||
auto &varRequestIdQueue = varIdToVarRequestIdQueueMapIt->second; | ||||
varRequestIdQueue.erase( | ||||
std::remove(varRequestIdQueue.begin(), varRequestIdQueue.end(), varRequestId), | ||||
varRequestIdQueue.end()); | ||||
if (varRequestIdQueue.empty()) { | ||||
varIdToVarRequestIdQueueMapIt | ||||
= m_VarIdToVarRequestIdQueueMap.erase(varIdToVarRequestIdQueueMapIt); | ||||
} | ||||
else { | ||||
++varIdToVarRequestIdQueueMapIt; | ||||
} | ||||
} | ||||
} | ||||