VCTransaction.h
148 lines
| 3.6 KiB
| text/x-c
|
CLexer
| r17 | #include "Variable/VariableSynchronizationGroup2.h" | |||
| r67 | ||||
| r17 | #include <Common/containers.h> | |||
| #include <Common/debug.h> | ||||
| #include <Data/DataProviderParameters.h> | ||||
| #include <Data/DateTimeRange.h> | ||||
| r67 | #include <Data/DateTimeRangeHelper.h> | |||
| r17 | #include <Data/IDataProvider.h> | |||
| r67 | #include <QObject> | |||
| #include <QReadWriteLock> | ||||
| #include <QRunnable> | ||||
| #include <QThreadPool> | ||||
| #include <TimeSeries.h> | ||||
| #include <Variable/Variable2.h> | ||||
| r17 | ||||
| struct VCTransaction | ||||
| { | ||||
| r67 | VCTransaction(QUuid refVar, DateTimeRange range, int varCount) | |||
| : refVar{refVar}, range{range}, _remainingVars{varCount} | ||||
| {} | ||||
| r19 | ||||
| r67 | QUuid refVar; | |||
| DateTimeRange range; | ||||
| bool ready() | ||||
| { | ||||
| QReadLocker lock{&_lock}; | ||||
| return _remainingVars == 0; | ||||
| } | ||||
| bool done() | ||||
| { | ||||
| QWriteLocker lock{&_lock}; | ||||
| _remainingVars -= 1; | ||||
| return _remainingVars == 0; | ||||
| } | ||||
| r19 | ||||
| private: | ||||
| r67 | QReadWriteLock _lock; | |||
| int _remainingVars; | ||||
| r17 | }; | |||
| r67 | class TransactionExe : public QObject, public QRunnable | |||
| r17 | { | |||
| r67 | Q_OBJECT | |||
| std::shared_ptr<Variable2> _variable; | ||||
| std::shared_ptr<IDataProvider> _provider; | ||||
| std::vector<DateTimeRange> _ranges; | ||||
| DateTimeRange _range; | ||||
| bool _overwrite; | ||||
| r17 | public: | |||
| r67 | TransactionExe(const std::shared_ptr<Variable2>& variable, | |||
| const std::shared_ptr<IDataProvider>& provider, | ||||
| const std::vector<DateTimeRange>& ranges, DateTimeRange range, | ||||
| bool overwrite = true) | ||||
| : _variable{variable}, _provider{provider}, _ranges{ranges}, | ||||
| _range{range}, _overwrite{overwrite} | ||||
| { | ||||
| setAutoDelete(true); | ||||
| } | ||||
| void run() override | ||||
| { | ||||
| std::vector<TimeSeries::ITimeSerie*> data; | ||||
| for(auto range : _ranges) | ||||
| r17 | { | |||
| r67 | auto ds = _provider->getData( | |||
| DataProviderParameters{{range}, _variable->metadata()}); | ||||
| if(ds) data.push_back(ds); | ||||
| r17 | } | |||
| r67 | if(_overwrite) | |||
| _variable->setData(data, _range, true); | ||||
| else | ||||
| r17 | { | |||
| r68 | data.push_back(_variable->data().get()); | |||
| r67 | _variable->setData(data, _range, true); | |||
| r17 | } | |||
| r73 | std::for_each(std::begin(data), std::end(data), | |||
| [](TimeSeries::ITimeSerie* ts) { delete ts; }); | ||||
| r67 | emit transactionComplete(); | |||
| } | ||||
| r17 | signals: | |||
| r67 | void transactionComplete(); | |||
| r17 | }; | |||
| r20 | ||||
| class VCTransactionsQueues | ||||
| { | ||||
| r67 | QReadWriteLock _mutex{QReadWriteLock::Recursive}; | |||
| std::map<QUuid, std::optional<std::shared_ptr<VCTransaction>>> | ||||
| _nextTransactions; | ||||
| std::map<QUuid, std::optional<std::shared_ptr<VCTransaction>>> | ||||
| _pendingTransactions; | ||||
| r20 | public: | |||
| r67 | void addEntry(QUuid id) | |||
| { | ||||
| QWriteLocker lock{&_mutex}; | ||||
| _nextTransactions[id] = std::nullopt; | ||||
| _pendingTransactions[id] = std::nullopt; | ||||
| } | ||||
| r20 | ||||
| r67 | void removeEntry(QUuid id) | |||
| { | ||||
| QWriteLocker lock{&_mutex}; | ||||
| _nextTransactions.erase(id); | ||||
| _pendingTransactions.erase(id); | ||||
| } | ||||
| r20 | ||||
| r67 | std::map<QUuid, std::optional<std::shared_ptr<VCTransaction>>> | |||
| pendingTransactions() | ||||
| { | ||||
| QReadLocker lock{&_mutex}; | ||||
| return _pendingTransactions; | ||||
| } | ||||
| r20 | ||||
| r67 | std::map<QUuid, std::optional<std::shared_ptr<VCTransaction>>> | |||
| nextTransactions() | ||||
| { | ||||
| QReadLocker lock{&_mutex}; | ||||
| return _nextTransactions; | ||||
| } | ||||
| r20 | ||||
| r67 | std::optional<std::shared_ptr<VCTransaction>> start(QUuid id) | |||
| { | ||||
| QWriteLocker lock{&_mutex}; | ||||
| _pendingTransactions[id] = _nextTransactions[id]; | ||||
| _nextTransactions[id] = std::nullopt; | ||||
| return _pendingTransactions[id]; | ||||
| } | ||||
| r20 | ||||
| r67 | void enqueue(QUuid id, std::shared_ptr<VCTransaction> transaction) | |||
| { | ||||
| QWriteLocker lock{&_mutex}; | ||||
| _nextTransactions[id] = transaction; | ||||
| } | ||||
| r20 | ||||
| r67 | void complete(QUuid id) | |||
| { | ||||
| QWriteLocker lock{&_mutex}; | ||||
| _pendingTransactions[id] = std::nullopt; | ||||
| } | ||||
| r20 | ||||
| r67 | bool active(QUuid id) | |||
| { | ||||
| QReadLocker lock{&_mutex}; | ||||
| return _nextTransactions[id].has_value() || | ||||
| _pendingTransactions[id].has_value(); | ||||
| } | ||||
| r20 | }; | |||
