From 4d9746c828558c9f872e0aed52442863affb25d5 Mon Sep 17 00:00:00 2001 From: Christian Mollekopf Date: Thu, 3 Mar 2016 09:01:05 +0100 Subject: Fromatted the whole codebase with clang-format. clang-format -i */**{.cpp,.h} --- common/genericresource.cpp | 378 ++++++++++++++++++++++----------------------- 1 file changed, 188 insertions(+), 190 deletions(-) (limited to 'common/genericresource.cpp') diff --git a/common/genericresource.cpp b/common/genericresource.cpp index 74a8cfb..9c9a12f 100644 --- a/common/genericresource.cpp +++ b/common/genericresource.cpp @@ -20,7 +20,7 @@ #include static int sBatchSize = 100; -//This interval directly affects the roundtrip time of single commands +// This interval directly affects the roundtrip time of single commands static int sCommitInterval = 10; using namespace Sink; @@ -39,26 +39,23 @@ class ChangeReplay : public QObject { Q_OBJECT public: - typedef std::function(const QByteArray &type, const QByteArray &key, const QByteArray &value)> ReplayFunction; ChangeReplay(const QString &resourceName, const ReplayFunction &replayFunction) - : mStorage(storageLocation(), resourceName, Storage::ReadOnly), - mChangeReplayStore(storageLocation(), resourceName + ".changereplay", Storage::ReadWrite), - mReplayFunction(replayFunction) + : mStorage(storageLocation(), resourceName, Storage::ReadOnly), mChangeReplayStore(storageLocation(), resourceName + ".changereplay", Storage::ReadWrite), mReplayFunction(replayFunction) { - } qint64 getLastReplayedRevision() { qint64 lastReplayedRevision = 0; auto replayStoreTransaction = mChangeReplayStore.createTransaction(Storage::ReadOnly); - replayStoreTransaction.openDatabase().scan("lastReplayedRevision", [&lastReplayedRevision](const QByteArray &key, const QByteArray &value) -> bool { - lastReplayedRevision = value.toLongLong(); - return false; - }, [](const Storage::Error &) { - }); + replayStoreTransaction.openDatabase().scan("lastReplayedRevision", + [&lastReplayedRevision](const QByteArray &key, const QByteArray &value) -> bool { + lastReplayedRevision = value.toLongLong(); + return false; + }, + [](const Storage::Error &) {}); return lastReplayedRevision; } @@ -79,28 +76,30 @@ public slots: auto mainStoreTransaction = mStorage.createTransaction(Storage::ReadOnly); auto replayStoreTransaction = mChangeReplayStore.createTransaction(Storage::ReadWrite); qint64 lastReplayedRevision = 1; - replayStoreTransaction.openDatabase().scan("lastReplayedRevision", [&lastReplayedRevision](const QByteArray &key, const QByteArray &value) -> bool { - lastReplayedRevision = value.toLongLong(); - return false; - }, [](const Storage::Error &) { - }); + replayStoreTransaction.openDatabase().scan("lastReplayedRevision", + [&lastReplayedRevision](const QByteArray &key, const QByteArray &value) -> bool { + lastReplayedRevision = value.toLongLong(); + return false; + }, + [](const Storage::Error &) {}); const qint64 topRevision = Storage::maxRevision(mainStoreTransaction); Trace() << "Changereplay from " << lastReplayedRevision << " to " << topRevision; if (lastReplayedRevision <= topRevision) { qint64 revision = lastReplayedRevision; - for (;revision <= topRevision; revision++) { + for (; revision <= topRevision; revision++) { const auto uid = Storage::getUidFromRevision(mainStoreTransaction, revision); const auto type = Storage::getTypeFromRevision(mainStoreTransaction, revision); const auto key = Storage::assembleKey(uid, revision); - Storage::mainDatabase(mainStoreTransaction, type).scan(key, [&lastReplayedRevision, type, this](const QByteArray &key, const QByteArray &value) -> bool { - mReplayFunction(type, key, value).exec(); - //TODO make for loop async, and pass to async replay function together with type - Trace() << "Replaying " << key; - return false; - }, [key](const Storage::Error &) { - ErrorMsg() << "Failed to replay change " << key; - }); + Storage::mainDatabase(mainStoreTransaction, type) + .scan(key, + [&lastReplayedRevision, type, this](const QByteArray &key, const QByteArray &value) -> bool { + mReplayFunction(type, key, value).exec(); + // TODO make for loop async, and pass to async replay function together with type + Trace() << "Replaying " << key; + return false; + }, + [key](const Storage::Error &) { ErrorMsg() << "Failed to replay change " << key; }); } revision--; replayStoreTransaction.openDatabase().write("lastReplayedRevision", QByteArray::number(revision)); @@ -126,15 +125,12 @@ class CommandProcessor : public QObject { Q_OBJECT typedef std::function(void const *, size_t)> InspectionFunction; + public: - CommandProcessor(Sink::Pipeline *pipeline, QList commandQueues) - : QObject(), - mPipeline(pipeline), - mCommandQueues(commandQueues), - mProcessingLock(false) + CommandProcessor(Sink::Pipeline *pipeline, QList commandQueues) : QObject(), mPipeline(pipeline), mCommandQueues(commandQueues), mProcessingLock(false) { mPipeline->startTransaction(); - //FIXME Should be initialized to the current value of the change replay queue + // FIXME Should be initialized to the current value of the change replay queue mLowerBoundRevision = Storage::maxRevision(mPipeline->transaction()); mPipeline->commit(); @@ -176,18 +172,20 @@ private slots: return; } mProcessingLock = true; - auto job = processPipeline().then([this]() { - mProcessingLock = false; - if (messagesToProcessAvailable()) { - process(); - } - }).exec(); + auto job = processPipeline() + .then([this]() { + mProcessingLock = false; + if (messagesToProcessAvailable()) { + process(); + } + }) + .exec(); } KAsync::Job processQueuedCommand(const Sink::QueuedCommand *queuedCommand) { Log() << "Processing command: " << Sink::Commands::name(queuedCommand->commandId()); - //Throw command into appropriate pipeline + // Throw command into appropriate pipeline switch (queuedCommand->commandId()) { case Sink::Commands::DeleteEntityCommand: return mPipeline->deletedEntity(queuedCommand->command()->Data(), queuedCommand->command()->size()); @@ -197,9 +195,7 @@ private slots: return mPipeline->newEntity(queuedCommand->command()->Data(), queuedCommand->command()->size()); case Sink::Commands::InspectionCommand: if (mInspect) { - return mInspect(queuedCommand->command()->Data(), queuedCommand->command()->size()).then([]() { - return -1; - }); + return mInspect(queuedCommand->command()->Data(), queuedCommand->command()->size()).then([]() { return -1; }); } else { return KAsync::error(-1, "Missing inspection command."); } @@ -218,50 +214,47 @@ private slots: auto queuedCommand = Sink::GetQueuedCommand(data.constData()); const auto commandId = queuedCommand->commandId(); Trace() << "Dequeued Command: " << Sink::Commands::name(commandId); - return processQueuedCommand(queuedCommand).then( - [commandId](qint64 createdRevision) -> qint64 { - Trace() << "Command pipeline processed: " << Sink::Commands::name(commandId); - return createdRevision; - } - , - [](int errorCode, QString errorMessage) { - //FIXME propagate error, we didn't handle it - Warning() << "Error while processing queue command: " << errorMessage; - } - ); + return processQueuedCommand(queuedCommand) + .then( + [commandId](qint64 createdRevision) -> qint64 { + Trace() << "Command pipeline processed: " << Sink::Commands::name(commandId); + return createdRevision; + }, + [](int errorCode, QString errorMessage) { + // FIXME propagate error, we didn't handle it + Warning() << "Error while processing queue command: " << errorMessage; + }); } - //Process all messages of this queue + // Process all messages of this queue KAsync::Job processQueue(MessageQueue *queue) { auto time = QSharedPointer::create(); - return KAsync::start([this](){ - mPipeline->startTransaction(); - }).then(KAsync::dowhile( - [queue]() { return !queue->isEmpty(); }, - [this, queue, time](KAsync::Future &future) { - queue->dequeueBatch(sBatchSize, [this, time](const QByteArray &data) { - time->start(); - return KAsync::start([this, data, time](KAsync::Future &future) { - processQueuedCommand(data).then([&future, this, time](qint64 createdRevision) { - Trace() << "Created revision " << createdRevision << ". Processing took: " << Log::TraceTime(time->elapsed()); + return KAsync::start([this]() { mPipeline->startTransaction(); }) + .then(KAsync::dowhile([queue]() { return !queue->isEmpty(); }, + [this, queue, time](KAsync::Future &future) { + queue->dequeueBatch(sBatchSize, + [this, time](const QByteArray &data) { + time->start(); + return KAsync::start([this, data, time](KAsync::Future &future) { + processQueuedCommand(data) + .then([&future, this, time](qint64 createdRevision) { + Trace() << "Created revision " << createdRevision << ". Processing took: " << Log::TraceTime(time->elapsed()); + future.setFinished(); + }) + .exec(); + }); + }) + .then([&future, queue]() { future.setFinished(); }, + [&future](int i, QString error) { + if (i != MessageQueue::ErrorCodes::NoMessageFound) { + Warning() << "Error while getting message from messagequeue: " << error; + } future.setFinished(); - }).exec(); - }); - } - ).then([&future, queue](){ - future.setFinished(); - }, - [&future](int i, QString error) { - if (i != MessageQueue::ErrorCodes::NoMessageFound) { - Warning() << "Error while getting message from messagequeue: " << error; - } - future.setFinished(); - }).exec(); - } - )).then([this]() { - mPipeline->commit(); - }); + }) + .exec(); + })) + .then([this]() { mPipeline->commit(); }); } KAsync::Job processPipeline() @@ -276,29 +269,29 @@ private slots: mPipeline->commit(); Trace() << "Cleanup done." << Log::TraceTime(time->elapsed()); - //Go through all message queues - auto it = QSharedPointer >::create(mCommandQueues); - return KAsync::dowhile( - [it]() { return it->hasNext(); }, + // Go through all message queues + auto it = QSharedPointer>::create(mCommandQueues); + return KAsync::dowhile([it]() { return it->hasNext(); }, [it, this](KAsync::Future &future) { auto time = QSharedPointer::create(); time->start(); auto queue = it->next(); - processQueue(queue).then([&future, time]() { - Trace() << "Queue processed." << Log::TraceTime(time->elapsed()); - future.setFinished(); - }).exec(); - } - ); + processQueue(queue) + .then([&future, time]() { + Trace() << "Queue processed." << Log::TraceTime(time->elapsed()); + future.setFinished(); + }) + .exec(); + }); } private: Sink::Pipeline *mPipeline; - //Ordered by priority - QList mCommandQueues; + // Ordered by priority + QList mCommandQueues; bool mProcessingLock; - //The lowest revision we no longer need + // The lowest revision we no longer need qint64 mLowerBoundRevision; InspectionFunction mInspect; }; @@ -308,14 +301,14 @@ private: GenericResource::GenericResource(const QByteArray &resourceInstanceIdentifier, const QSharedPointer &pipeline) : Sink::Resource(), - mUserQueue(Sink::storageLocation(), resourceInstanceIdentifier + ".userqueue"), - mSynchronizerQueue(Sink::storageLocation(), resourceInstanceIdentifier + ".synchronizerqueue"), - mResourceInstanceIdentifier(resourceInstanceIdentifier), - mPipeline(pipeline ? pipeline : QSharedPointer::create(resourceInstanceIdentifier)), - mError(0), - mClientLowerBoundRevision(std::numeric_limits::max()) -{ - mProcessor = new CommandProcessor(mPipeline.data(), QList() << &mUserQueue << &mSynchronizerQueue); + mUserQueue(Sink::storageLocation(), resourceInstanceIdentifier + ".userqueue"), + mSynchronizerQueue(Sink::storageLocation(), resourceInstanceIdentifier + ".synchronizerqueue"), + mResourceInstanceIdentifier(resourceInstanceIdentifier), + mPipeline(pipeline ? pipeline : QSharedPointer::create(resourceInstanceIdentifier)), + mError(0), + mClientLowerBoundRevision(std::numeric_limits::max()) +{ + mProcessor = new CommandProcessor(mPipeline.data(), QList() << &mUserQueue << &mSynchronizerQueue); mProcessor->setInspectionCommand([this](void const *command, size_t size) { flatbuffers::Verifier verifier((const uint8_t *)command, size); if (Sink::Commands::VerifyInspectionBuffer(verifier)) { @@ -330,22 +323,26 @@ GenericResource::GenericResource(const QByteArray &resourceInstanceIdentifier, c QDataStream s(expectedValueString); QVariant expectedValue; s >> expectedValue; - inspect(inspectionType, inspectionId, domainType, entityId, property, expectedValue).then([=]() { - Log_area("resource.inspection") << "Inspection was successful: " << inspectionType << inspectionId << entityId; - Sink::Notification n; - n.type = Sink::Commands::NotificationType_Inspection; - n.id = inspectionId; - n.code = Sink::Commands::NotificationCode_Success; - emit notify(n); - }, [=](int code, const QString &message) { - Log() << "Inspection failed: "<< inspectionType << inspectionId << entityId << message; - Sink::Notification n; - n.type = Sink::Commands::NotificationType_Inspection; - n.message = message; - n.id = inspectionId; - n.code = Sink::Commands::NotificationCode_Failure; - emit notify(n); - }).exec(); + inspect(inspectionType, inspectionId, domainType, entityId, property, expectedValue) + .then( + [=]() { + Log_area("resource.inspection") << "Inspection was successful: " << inspectionType << inspectionId << entityId; + Sink::Notification n; + n.type = Sink::Commands::NotificationType_Inspection; + n.id = inspectionId; + n.code = Sink::Commands::NotificationCode_Success; + emit notify(n); + }, + [=](int code, const QString &message) { + Log() << "Inspection failed: " << inspectionType << inspectionId << entityId << message; + Sink::Notification n; + n.type = Sink::Commands::NotificationType_Inspection; + n.message = message; + n.id = inspectionId; + n.code = Sink::Commands::NotificationCode_Failure; + emit notify(n); + }) + .exec(); return KAsync::null(); } return KAsync::error(-1, "Invalid inspection command."); @@ -353,9 +350,9 @@ GenericResource::GenericResource(const QByteArray &resourceInstanceIdentifier, c QObject::connect(mProcessor, &CommandProcessor::error, [this](int errorCode, const QString &msg) { onProcessorError(errorCode, msg); }); QObject::connect(mPipeline.data(), &Pipeline::revisionUpdated, this, &Resource::revisionUpdated); mSourceChangeReplay = new ChangeReplay(resourceInstanceIdentifier, [this](const QByteArray &type, const QByteArray &key, const QByteArray &value) { - //This results in a deadlock when a sync is in progress and we try to create a second writing transaction (which is why we turn changereplay off during the sync) + // This results in a deadlock when a sync is in progress and we try to create a second writing transaction (which is why we turn changereplay off during the sync) auto synchronizationStore = QSharedPointer::create(Sink::storageLocation(), mResourceInstanceIdentifier + ".synchronization", Sink::Storage::ReadWrite); - return this->replay(*synchronizationStore, type, key, value).then([synchronizationStore](){}); + return this->replay(*synchronizationStore, type, key, value).then([synchronizationStore]() {}); }); enableChangeReplay(true); mClientLowerBoundRevision = mPipeline->cleanedUpRevision(); @@ -372,7 +369,8 @@ GenericResource::~GenericResource() delete mSourceChangeReplay; } -KAsync::Job GenericResource::inspect(int inspectionType, const QByteArray &inspectionId, const QByteArray &domainType, const QByteArray &entityId, const QByteArray &property, const QVariant &expectedValue) +KAsync::Job GenericResource::inspect( + int inspectionType, const QByteArray &inspectionId, const QByteArray &domainType, const QByteArray &entityId, const QByteArray &property, const QVariant &expectedValue) { Warning() << "Inspection not implemented"; return KAsync::null(); @@ -390,7 +388,7 @@ void GenericResource::enableChangeReplay(bool enable) } } -void GenericResource::addType(const QByteArray &type, DomainTypeAdaptorFactoryInterface::Ptr factory, const QVector &preprocessors) +void GenericResource::addType(const QByteArray &type, DomainTypeAdaptorFactoryInterface::Ptr factory, const QVector &preprocessors) { mPipeline->setPreprocessors(type, preprocessors); mPipeline->setAdaptorFactory(type, factory); @@ -463,14 +461,16 @@ KAsync::Job GenericResource::synchronizeWithSource() { return KAsync::start([this]() { Log() << " Synchronizing"; - //Changereplay would deadlock otherwise when trying to open the synchronization store + // Changereplay would deadlock otherwise when trying to open the synchronization store enableChangeReplay(false); auto mainStore = QSharedPointer::create(Sink::storageLocation(), mResourceInstanceIdentifier, Sink::Storage::ReadOnly); auto syncStore = QSharedPointer::create(Sink::storageLocation(), mResourceInstanceIdentifier + ".synchronization", Sink::Storage::ReadWrite); - synchronizeWithSource(*mainStore, *syncStore).then([this, mainStore, syncStore]() { - Log() << "Done Synchronizing"; - enableChangeReplay(true); - }).exec(); + synchronizeWithSource(*mainStore, *syncStore) + .then([this, mainStore, syncStore]() { + Log() << "Done Synchronizing"; + enableChangeReplay(true); + }) + .exec(); }); } @@ -484,42 +484,39 @@ static void waitForDrained(KAsync::Future &f, MessageQueue &queue) if (queue.isEmpty()) { f.setFinished(); } else { - QObject::connect(&queue, &MessageQueue::drained, [&f]() { - f.setFinished(); - }); + QObject::connect(&queue, &MessageQueue::drained, [&f]() { f.setFinished(); }); } }; KAsync::Job GenericResource::processAllMessages() { - //We have to wait for all items to be processed to ensure the synced items are available when a query gets executed. - //TODO: report errors while processing sync? - //TODO JOBAPI: A helper that waits for n events and then continues? + // We have to wait for all items to be processed to ensure the synced items are available when a query gets executed. + // TODO: report errors while processing sync? + // TODO JOBAPI: A helper that waits for n events and then continues? return KAsync::start([this](KAsync::Future &f) { - if (mCommitQueueTimer.isActive()) { - auto context = new QObject; - QObject::connect(&mCommitQueueTimer, &QTimer::timeout, context, [&f, context]() { - delete context; + if (mCommitQueueTimer.isActive()) { + auto context = new QObject; + QObject::connect(&mCommitQueueTimer, &QTimer::timeout, context, [&f, context]() { + delete context; + f.setFinished(); + }); + } else { + f.setFinished(); + } + }) + .then([this](KAsync::Future &f) { waitForDrained(f, mSynchronizerQueue); }) + .then([this](KAsync::Future &f) { waitForDrained(f, mUserQueue); }) + .then([this](KAsync::Future &f) { + if (mSourceChangeReplay->allChangesReplayed()) { f.setFinished(); - }); - } else { - f.setFinished(); - } - }).then([this](KAsync::Future &f) { - waitForDrained(f, mSynchronizerQueue); - }).then([this](KAsync::Future &f) { - waitForDrained(f, mUserQueue); - }).then([this](KAsync::Future &f) { - if (mSourceChangeReplay->allChangesReplayed()) { - f.setFinished(); - } else { - auto context = new QObject; - QObject::connect(mSourceChangeReplay, &ChangeReplay::changesReplayed, context, [&f, context]() { - delete context; - f.setFinished(); - }); - } - }); + } else { + auto context = new QObject; + QObject::connect(mSourceChangeReplay, &ChangeReplay::changesReplayed, context, [&f, context]() { + delete context; + f.setFinished(); + }); + } + }); } void GenericResource::updateLowerBoundRevision() @@ -533,14 +530,15 @@ void GenericResource::setLowerBoundRevision(qint64 revision) updateLowerBoundRevision(); } -void GenericResource::createEntity(const QByteArray &sinkId, const QByteArray &bufferType, const Sink::ApplicationDomain::ApplicationDomainType &domainObject, DomainTypeAdaptorFactoryInterface &adaptorFactory, std::function callback) +void GenericResource::createEntity(const QByteArray &sinkId, const QByteArray &bufferType, const Sink::ApplicationDomain::ApplicationDomainType &domainObject, + DomainTypeAdaptorFactoryInterface &adaptorFactory, std::function callback) { - //These changes are coming from the source + // These changes are coming from the source const auto replayToSource = false; flatbuffers::FlatBufferBuilder entityFbb; adaptorFactory.createBuffer(domainObject, entityFbb); flatbuffers::FlatBufferBuilder fbb; - //This is the resource type and not the domain type + // This is the resource type and not the domain type auto entityId = fbb.CreateString(sinkId.toStdString()); auto type = fbb.CreateString(bufferType.toStdString()); auto delta = Sink::EntityBuffer::appendAsVector(fbb, entityFbb.GetBufferPointer(), entityFbb.GetSize()); @@ -549,18 +547,19 @@ void GenericResource::createEntity(const QByteArray &sinkId, const QByteArray &b callback(BufferUtils::extractBuffer(fbb)); } -void GenericResource::modifyEntity(const QByteArray &sinkId, qint64 revision, const QByteArray &bufferType, const Sink::ApplicationDomain::ApplicationDomainType &domainObject, DomainTypeAdaptorFactoryInterface &adaptorFactory, std::function callback) +void GenericResource::modifyEntity(const QByteArray &sinkId, qint64 revision, const QByteArray &bufferType, const Sink::ApplicationDomain::ApplicationDomainType &domainObject, + DomainTypeAdaptorFactoryInterface &adaptorFactory, std::function callback) { - //These changes are coming from the source + // These changes are coming from the source const auto replayToSource = false; flatbuffers::FlatBufferBuilder entityFbb; adaptorFactory.createBuffer(domainObject, entityFbb); flatbuffers::FlatBufferBuilder fbb; auto entityId = fbb.CreateString(sinkId.toStdString()); - //This is the resource type and not the domain type + // This is the resource type and not the domain type auto type = fbb.CreateString(bufferType.toStdString()); auto delta = Sink::EntityBuffer::appendAsVector(fbb, entityFbb.GetBufferPointer(), entityFbb.GetSize()); - //TODO removals + // TODO removals auto location = Sink::Commands::CreateModifyEntity(fbb, revision, entityId, 0, type, delta, replayToSource); Sink::Commands::FinishModifyEntityBuffer(fbb, location); callback(BufferUtils::extractBuffer(fbb)); @@ -568,11 +567,11 @@ void GenericResource::modifyEntity(const QByteArray &sinkId, qint64 revision, co void GenericResource::deleteEntity(const QByteArray &sinkId, qint64 revision, const QByteArray &bufferType, std::function callback) { - //These changes are coming from the source + // These changes are coming from the source const auto replayToSource = false; flatbuffers::FlatBufferBuilder fbb; auto entityId = fbb.CreateString(sinkId.toStdString()); - //This is the resource type and not the domain type + // This is the resource type and not the domain type auto type = fbb.CreateString(bufferType.toStdString()); auto location = Sink::Commands::CreateDeleteEntity(fbb, revision, entityId, type, replayToSource); Sink::Commands::FinishDeleteEntityBuffer(fbb, location); @@ -581,7 +580,8 @@ void GenericResource::deleteEntity(const QByteArray &sinkId, qint64 revision, co void GenericResource::recordRemoteId(const QByteArray &bufferType, const QByteArray &localId, const QByteArray &remoteId, Sink::Storage::Transaction &transaction) { - Index("rid.mapping." + bufferType, transaction).add(remoteId, localId);; + Index("rid.mapping." + bufferType, transaction).add(remoteId, localId); + ; Index("localid.mapping." + bufferType, transaction).add(localId, remoteId); } @@ -600,7 +600,7 @@ void GenericResource::updateRemoteId(const QByteArray &bufferType, const QByteAr QByteArray GenericResource::resolveRemoteId(const QByteArray &bufferType, const QByteArray &remoteId, Sink::Storage::Transaction &transaction) { - //Lookup local id for remote id, or insert a new pair otherwise + // Lookup local id for remote id, or insert a new pair otherwise Index index("rid.mapping." + bufferType, transaction); QByteArray sinkId = index.lookup(remoteId); if (sinkId.isEmpty()) { @@ -621,19 +621,19 @@ QByteArray GenericResource::resolveLocalId(const QByteArray &bufferType, const Q return remoteId; } -void GenericResource::scanForRemovals(Sink::Storage::Transaction &transaction, Sink::Storage::Transaction &synchronizationTransaction, const QByteArray &bufferType, const std::function &callback)> &entryGenerator, std::function exists) +void GenericResource::scanForRemovals(Sink::Storage::Transaction &transaction, Sink::Storage::Transaction &synchronizationTransaction, const QByteArray &bufferType, + const std::function &callback)> &entryGenerator, std::function exists) { entryGenerator([this, &transaction, bufferType, &synchronizationTransaction, &exists](const QByteArray &key) { auto sinkId = Sink::Storage::uidFromKey(key); Trace() << "Checking for removal " << key; const auto remoteId = resolveLocalId(bufferType, sinkId, synchronizationTransaction); - //If we have no remoteId, the entity hasn't been replayed to the source yet + // If we have no remoteId, the entity hasn't been replayed to the source yet if (!remoteId.isEmpty()) { if (!exists(remoteId)) { Trace() << "Found a removed entity: " << sinkId; - deleteEntity(sinkId, Sink::Storage::maxRevision(transaction), bufferType, [this](const QByteArray &buffer) { - enqueueCommand(mSynchronizerQueue, Sink::Commands::DeleteEntityCommand, buffer); - }); + deleteEntity(sinkId, Sink::Storage::maxRevision(transaction), bufferType, + [this](const QByteArray &buffer) { enqueueCommand(mSynchronizerQueue, Sink::Commands::DeleteEntityCommand, buffer); }); } } }); @@ -642,32 +642,31 @@ void GenericResource::scanForRemovals(Sink::Storage::Transaction &transaction, S static QSharedPointer getLatest(const Sink::Storage::NamedDatabase &db, const QByteArray &uid, DomainTypeAdaptorFactoryInterface &adaptorFactory) { QSharedPointer current; - db.findLatest(uid, [¤t, &adaptorFactory](const QByteArray &key, const QByteArray &data) -> bool { - Sink::EntityBuffer buffer(const_cast(data.data()), data.size()); - if (!buffer.isValid()) { - Warning() << "Read invalid buffer from disk"; - } else { - current = adaptorFactory.createAdaptor(buffer.entity()); - } - return false; - }, - [](const Sink::Storage::Error &error) { - Warning() << "Failed to read current value from storage: " << error.message; - }); + db.findLatest(uid, + [¤t, &adaptorFactory](const QByteArray &key, const QByteArray &data) -> bool { + Sink::EntityBuffer buffer(const_cast(data.data()), data.size()); + if (!buffer.isValid()) { + Warning() << "Read invalid buffer from disk"; + } else { + current = adaptorFactory.createAdaptor(buffer.entity()); + } + return false; + }, + [](const Sink::Storage::Error &error) { Warning() << "Failed to read current value from storage: " << error.message; }); return current; } -void GenericResource::createOrModify(Sink::Storage::Transaction &transaction, Sink::Storage::Transaction &synchronizationTransaction, DomainTypeAdaptorFactoryInterface &adaptorFactory, const QByteArray &bufferType, const QByteArray &remoteId, const Sink::ApplicationDomain::ApplicationDomainType &entity) +void GenericResource::createOrModify(Sink::Storage::Transaction &transaction, Sink::Storage::Transaction &synchronizationTransaction, + DomainTypeAdaptorFactoryInterface &adaptorFactory, const QByteArray &bufferType, const QByteArray &remoteId, const Sink::ApplicationDomain::ApplicationDomainType &entity) { auto mainDatabase = Storage::mainDatabase(transaction, bufferType); const auto sinkId = resolveRemoteId(bufferType, remoteId, synchronizationTransaction); const auto found = mainDatabase.contains(sinkId); if (!found) { Trace() << "Found a new entity: " << remoteId; - createEntity(sinkId, bufferType, entity, adaptorFactory, [this](const QByteArray &buffer) { - enqueueCommand(mSynchronizerQueue, Sink::Commands::CreateEntityCommand, buffer); - }); - } else { //modification + createEntity( + sinkId, bufferType, entity, adaptorFactory, [this](const QByteArray &buffer) { enqueueCommand(mSynchronizerQueue, Sink::Commands::CreateEntityCommand, buffer); }); + } else { // modification if (auto current = getLatest(mainDatabase, sinkId, adaptorFactory)) { bool changed = false; for (const auto &property : entity.changedProperties()) { @@ -678,9 +677,8 @@ void GenericResource::createOrModify(Sink::Storage::Transaction &transaction, Si } if (changed) { Trace() << "Found a modified entity: " << remoteId; - modifyEntity(sinkId, Sink::Storage::maxRevision(transaction), bufferType, entity, adaptorFactory, [this](const QByteArray &buffer) { - enqueueCommand(mSynchronizerQueue, Sink::Commands::ModifyEntityCommand, buffer); - }); + modifyEntity(sinkId, Sink::Storage::maxRevision(transaction), bufferType, entity, adaptorFactory, + [this](const QByteArray &buffer) { enqueueCommand(mSynchronizerQueue, Sink::Commands::ModifyEntityCommand, buffer); }); } } else { Warning() << "Failed to get current entity"; -- cgit v1.2.3