1 files changed, 53 insertions, 24 deletions

diff --git a/common/genericresource.cpp b/common/genericresource.cpp
index c7326d3..2688df0 100644
--- a/common/genericresource.cpp
+++ b/common/genericresource.cpp
@@ -17,11 +17,17 @@
 
 #include <QUuid>
 #include <QDataStream>
+#include <QTime>
 
 static int sBatchSize = 100;
+//This interval directly affects the roundtrip time of single commands
+static int sCommitInterval = 10;
 
 using namespace Sink;
 
+#undef DEBUG_AREA
+#define DEBUG_AREA "resource.changereplay"
+
 /**
  * Replays changes from the storage one by one.
  *
@@ -87,7 +93,7 @@ public Q_SLOTS:
                 const auto uid = Storage::getUidFromRevision(mainStoreTransaction, revision);
                 const auto type = Storage::getTypeFromRevision(mainStoreTransaction, revision);
                 const auto key = Storage::assembleKey(uid, revision);
-                mainStoreTransaction.openDatabase(type + ".main").scan(key, [&lastReplayedRevision, type, this](const QByteArray &key, const QByteArray &value) -> bool {
+                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;
@@ -110,6 +116,9 @@ private:
     ReplayFunction mReplayFunction;
 };
 
+#undef DEBUG_AREA
+#define DEBUG_AREA "resource.commandprocessor"
+
 /**
  * Drives the pipeline using the output from all command queues
  */
@@ -197,7 +206,6 @@ private slots:
             default:
                 return KAsync::error<qint64>(-1, "Unhandled command");
         }
-        return KAsync::null<qint64>();
     }
 
     KAsync::Job<qint64, qint64> processQueuedCommand(const QByteArray &data)
@@ -226,15 +234,17 @@ private slots:
     //Process all messages of this queue
     KAsync::Job<void> processQueue(MessageQueue *queue)
     {
+        auto time = QSharedPointer<QTime>::create();
         return KAsync::start<void>([this](){
             mPipeline->startTransaction();
         }).then(KAsync::dowhile(
             [queue]() { return !queue->isEmpty(); },
-            [this, queue](KAsync::Future<void> &future) {
-                queue->dequeueBatch(sBatchSize, [this](const QByteArray &data) {
-                        return KAsync::start<void>([this, data](KAsync::Future<void> &future) {
-                            processQueuedCommand(data).then<void, qint64>([&future, this](qint64 createdRevision) {
-                                Trace() << "Created revision " << createdRevision;
+            [this, queue, time](KAsync::Future<void> &future) {
+                queue->dequeueBatch(sBatchSize, [this, time](const QByteArray &data) {
+                        time->start();
+                        return KAsync::start<void>([this, data, time](KAsync::Future<void> &future) {
+                            processQueuedCommand(data).then<void, qint64>([&future, this, time](qint64 createdRevision) {
+                                Trace() << "Created revision " << createdRevision << ". Processing took: " << Log::TraceTime(time->elapsed());
                                 future.setFinished();
                             }).exec();
                         });
@@ -256,21 +266,27 @@ private slots:
 
     KAsync::Job<void> processPipeline()
     {
+        auto time = QSharedPointer<QTime>::create();
+        time->start();
         mPipeline->startTransaction();
         Trace() << "Cleaning up from " << mPipeline->cleanedUpRevision() + 1 << " to " << mLowerBoundRevision;
         for (qint64 revision = mPipeline->cleanedUpRevision() + 1; revision <= mLowerBoundRevision; revision++) {
             mPipeline->cleanupRevision(revision);
         }
         mPipeline->commit();
+        Trace() << "Cleanup done." << Log::TraceTime(time->elapsed());
 
         //Go through all message queues
         auto it = QSharedPointer<QListIterator<MessageQueue*> >::create(mCommandQueues);
         return KAsync::dowhile(
             [it]() { return it->hasNext(); },
             [it, this](KAsync::Future<void> &future) {
+                auto time = QSharedPointer<QTime>::create();
+                time->start();
+
                 auto queue = it->next();
-                processQueue(queue).then<void>([&future]() {
-                    Trace() << "Queue processed";
+                processQueue(queue).then<void>([&future, time]() {
+                    Trace() << "Queue processed." << Log::TraceTime(time->elapsed());
                     future.setFinished();
                 }).exec();
             }
@@ -287,6 +303,8 @@ private:
     InspectionFunction mInspect;
 };
 
+#undef DEBUG_AREA
+#define DEBUG_AREA "resource"
 
 GenericResource::GenericResource(const QByteArray &resourceInstanceIdentifier, const QSharedPointer<Pipeline> &pipeline)
     : Sink::Resource(),
@@ -313,12 +331,14 @@ GenericResource::GenericResource(const QByteArray &resourceInstanceIdentifier, c
             QVariant expectedValue;
             s >> expectedValue;
             inspect(inspectionType, inspectionId, domainType, entityId, property, expectedValue).then<void>([=]() {
+                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;
@@ -341,7 +361,7 @@ GenericResource::GenericResource(const QByteArray &resourceInstanceIdentifier, c
     mClientLowerBoundRevision = mPipeline->cleanedUpRevision();
     mProcessor->setOldestUsedRevision(mSourceChangeReplay->getLastReplayedRevision());
 
-    mCommitQueueTimer.setInterval(100);
+    mCommitQueueTimer.setInterval(sCommitInterval);
     mCommitQueueTimer.setSingleShot(true);
     QObject::connect(&mCommitQueueTimer, &QTimer::timeout, &mUserQueue, &MessageQueue::commit);
 }
@@ -381,13 +401,18 @@ KAsync::Job<void> GenericResource::replay(Sink::Storage &synchronizationStore, c
     return KAsync::null<void>();
 }
 
+void GenericResource::removeDataFromDisk()
+{
+    removeFromDisk(mResourceInstanceIdentifier);
+}
+
 void GenericResource::removeFromDisk(const QByteArray &instanceIdentifier)
 {
-    Warning() << "Removing from generic resource";
     Sink::Storage(Sink::storageLocation(), instanceIdentifier, Sink::Storage::ReadWrite).removeFromDisk();
     Sink::Storage(Sink::storageLocation(), instanceIdentifier + ".userqueue", Sink::Storage::ReadWrite).removeFromDisk();
     Sink::Storage(Sink::storageLocation(), instanceIdentifier + ".synchronizerqueue", Sink::Storage::ReadWrite).removeFromDisk();
     Sink::Storage(Sink::storageLocation(), instanceIdentifier + ".changereplay", Sink::Storage::ReadWrite).removeFromDisk();
+    Sink::Storage(Sink::storageLocation(), instanceIdentifier + ".synchronization", Sink::Storage::ReadWrite).removeFromDisk();
 }
 
 qint64 GenericResource::diskUsage(const QByteArray &instanceIdentifier)
@@ -556,38 +581,39 @@ 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 index("rid.mapping." + bufferType, transaction);
-    Index localIndex("localid.mapping." + bufferType, transaction);
-    index.add(remoteId, localId);
-    localIndex.add(localId, remoteId);
+    Index("rid.mapping." + bufferType, transaction).add(remoteId, localId);;
+    Index("localid.mapping." + bufferType, transaction).add(localId, remoteId);
 }
 
 void GenericResource::removeRemoteId(const QByteArray &bufferType, const QByteArray &localId, const QByteArray &remoteId, Sink::Storage::Transaction &transaction)
 {
-    Index index("rid.mapping." + bufferType, transaction);
-    Index localIndex("localid.mapping." + bufferType, transaction);
-    index.remove(remoteId, localId);
-    localIndex.remove(localId, remoteId);
+    Index("rid.mapping." + bufferType, transaction).remove(remoteId, localId);
+    Index("localid.mapping." + bufferType, transaction).remove(localId, remoteId);
+}
+
+void GenericResource::updateRemoteId(const QByteArray &bufferType, const QByteArray &localId, const QByteArray &remoteId, Sink::Storage::Transaction &transaction)
+{
+    const auto oldRemoteId = Index("localid.mapping." + bufferType, transaction).lookup(localId);
+    removeRemoteId(bufferType, localId, oldRemoteId, transaction);
+    recordRemoteId(bufferType, localId, remoteId, transaction);
 }
 
 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
     Index index("rid.mapping." + bufferType, transaction);
-    Index localIndex("localid.mapping." + bufferType, transaction);
     QByteArray sinkId = index.lookup(remoteId);
     if (sinkId.isEmpty()) {
         sinkId = QUuid::createUuid().toString().toUtf8();
         index.add(remoteId, sinkId);
-        localIndex.add(sinkId, remoteId);
+        Index("localid.mapping." + bufferType, transaction).add(sinkId, remoteId);
     }
     return sinkId;
 }
 
 QByteArray GenericResource::resolveLocalId(const QByteArray &bufferType, const QByteArray &localId, Sink::Storage::Transaction &transaction)
 {
-    Index index("localid.mapping." + bufferType, transaction);
-    QByteArray remoteId = index.lookup(localId);
+    QByteArray remoteId = Index("localid.mapping." + bufferType, transaction).lookup(localId);
     if (remoteId.isEmpty()) {
         Warning() << "Couldn't find the remote id for " << localId;
         return QByteArray();
@@ -633,7 +659,7 @@ static QSharedPointer<Sink::ApplicationDomain::BufferAdaptor> getLatest(const Si
 
 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 = transaction.openDatabase(bufferType + ".main");
+    auto mainDatabase = Storage::mainDatabase(transaction, bufferType);
     const auto sinkId = resolveRemoteId(bufferType, remoteId, synchronizationTransaction);
     const auto found = mainDatabase.contains(sinkId);
     if (!found) {
@@ -663,4 +689,7 @@ void GenericResource::createOrModify(Sink::Storage::Transaction &transaction, Si
 }
 
 
+#pragma clang diagnostic push
+#pragma clang diagnostic ignored "-Wundefined-reinterpret-cast"
 #include "genericresource.moc"
+#pragma clang diagnostic pop