Folded the SourceWriteback into the Synchronizer.

By concentrating all communication to the source in one place we get rid
of several oddities.
* Quite a bit of duplication since both need access to the
  synchronizationStore and the source.
* We currently have an akward locking in place because both classes
  access the ync store. This is not easier to resolve cleanly.
* The live of resource implementers becomes easier.
* An implementation could elect to not use changereplay and always do a
  full sync... (maybe?)

7 files changed, 120 insertions, 240 deletions

diff --git a/common/CMakeLists.txt b/common/CMakeLists.txt
index a08be8a..5ba524b 100644
--- a/common/CMakeLists.txt
+++ b/common/CMakeLists.txt
@@ -70,7 +70,6 @@ set(command_SRCS
     adaptorfactoryregistry.cpp
     synchronizer.cpp
     remoteidmap.cpp
-    sourcewriteback.cpp
     mailpreprocessor.cpp
     specialpurposepreprocessor.cpp
     datastorequery.cpp
diff --git a/common/genericresource.cpp b/common/genericresource.cpp
index c4c8bc6..746fa33 100644
--- a/common/genericresource.cpp
+++ b/common/genericresource.cpp
@@ -292,14 +292,14 @@ KAsync::Job<void> GenericResource::inspect(
 
 void GenericResource::enableChangeReplay(bool enable)
 {
-    Q_ASSERT(mChangeReplay);
+    Q_ASSERT(mSynchronizer);
     if (enable) {
-        QObject::connect(mPipeline.data(), &Pipeline::revisionUpdated, mChangeReplay.data(), &ChangeReplay::revisionChanged, Qt::QueuedConnection);
-        QObject::connect(mChangeReplay.data(), &ChangeReplay::changesReplayed, this, &GenericResource::updateLowerBoundRevision);
-        QMetaObject::invokeMethod(mChangeReplay.data(), "revisionChanged", Qt::QueuedConnection);
+        QObject::connect(mPipeline.data(), &Pipeline::revisionUpdated, mSynchronizer.data(), &ChangeReplay::revisionChanged, Qt::QueuedConnection);
+        QObject::connect(mSynchronizer.data(), &ChangeReplay::changesReplayed, this, &GenericResource::updateLowerBoundRevision);
+        QMetaObject::invokeMethod(mSynchronizer.data(), "revisionChanged", Qt::QueuedConnection);
     } else {
-        QObject::disconnect(mPipeline.data(), &Pipeline::revisionUpdated, mChangeReplay.data(), &ChangeReplay::revisionChanged);
-        QObject::disconnect(mChangeReplay.data(), &ChangeReplay::changesReplayed, this, &GenericResource::updateLowerBoundRevision);
+        QObject::disconnect(mPipeline.data(), &Pipeline::revisionUpdated, mSynchronizer.data(), &ChangeReplay::revisionChanged);
+        QObject::disconnect(mSynchronizer.data(), &ChangeReplay::changesReplayed, this, &GenericResource::updateLowerBoundRevision);
     }
 }
 
@@ -314,13 +314,8 @@ void GenericResource::setupSynchronizer(const QSharedPointer<Synchronizer> &sync
     mSynchronizer->setup([this](int commandId, const QByteArray &data) {
         enqueueCommand(mSynchronizerQueue, commandId, data);
     }, mSynchronizerQueue);
-}
-
-void GenericResource::setupChangereplay(const QSharedPointer<ChangeReplay> &changeReplay)
-{
-    mChangeReplay = changeReplay;
     {
-        auto ret = QObject::connect(mChangeReplay.data(), &ChangeReplay::replayingChanges, [this]() {
+        auto ret = QObject::connect(mSynchronizer.data(), &Synchronizer::replayingChanges, [this]() {
             Sink::Notification n;
             n.id = "changereplay";
             n.type = Sink::Notification::Status;
@@ -331,7 +326,7 @@ void GenericResource::setupChangereplay(const QSharedPointer<ChangeReplay> &chan
         Q_ASSERT(ret);
     }
     {
-        auto ret = QObject::connect(mChangeReplay.data(), &ChangeReplay::changesReplayed, [this]() {
+        auto ret = QObject::connect(mSynchronizer.data(), &Synchronizer::changesReplayed, [this]() {
             Sink::Notification n;
             n.id = "changereplay";
             n.type = Sink::Notification::Status;
@@ -342,7 +337,7 @@ void GenericResource::setupChangereplay(const QSharedPointer<ChangeReplay> &chan
         Q_ASSERT(ret);
     }
 
-    mProcessor->setOldestUsedRevision(mChangeReplay->getLastReplayedRevision());
+    mProcessor->setOldestUsedRevision(mSynchronizer->getLastReplayedRevision());
     enableChangeReplay(true);
 }
 
@@ -459,11 +454,11 @@ KAsync::Job<void> GenericResource::processAllMessages()
         .then<void>([this](KAsync::Future<void> &f) { waitForDrained(f, mSynchronizerQueue); })
         .then<void>([this](KAsync::Future<void> &f) { waitForDrained(f, mUserQueue); })
         .then<void>([this](KAsync::Future<void> &f) {
-            if (mChangeReplay->allChangesReplayed()) {
+            if (mSynchronizer->allChangesReplayed()) {
                 f.setFinished();
             } else {
                 auto context = new QObject;
-                QObject::connect(mChangeReplay.data(), &ChangeReplay::changesReplayed, context, [&f, context]() {
+                QObject::connect(mSynchronizer.data(), &ChangeReplay::changesReplayed, context, [&f, context]() {
                     delete context;
                     f.setFinished();
                 });
@@ -473,7 +468,7 @@ KAsync::Job<void> GenericResource::processAllMessages()
 
 void GenericResource::updateLowerBoundRevision()
 {
-    mProcessor->setOldestUsedRevision(qMin(mClientLowerBoundRevision, mChangeReplay->getLastReplayedRevision()));
+    mProcessor->setOldestUsedRevision(qMin(mClientLowerBoundRevision, mSynchronizer->getLastReplayedRevision()));
 }
 
 void GenericResource::setLowerBoundRevision(qint64 revision)
diff --git a/common/genericresource.h b/common/genericresource.h
index 3736c8f..7e0f5ad 100644
--- a/common/genericresource.h
+++ b/common/genericresource.h
@@ -24,7 +24,7 @@
 #include <messagequeue.h>
 #include <flatbuffers/flatbuffers.h>
 #include <domainadaptor.h>
-#include "changereplay.h"
+#include <resourcecontext.h>
 
 #include <QTimer>
 
@@ -66,7 +66,6 @@ protected:
 
     void setupPreprocessors(const QByteArray &type, const QVector<Sink::Preprocessor *> &preprocessors);
     void setupSynchronizer(const QSharedPointer<Synchronizer> &synchronizer);
-    void setupChangereplay(const QSharedPointer<ChangeReplay> &changeReplay);
 
     void onProcessorError(int errorCode, const QString &errorMessage);
     void enqueueCommand(MessageQueue &mq, int commandId, const QByteArray &data);
@@ -78,7 +77,6 @@ protected:
 
 private:
     std::unique_ptr<CommandProcessor> mProcessor;
-    QSharedPointer<ChangeReplay> mChangeReplay;
     QSharedPointer<Synchronizer> mSynchronizer;
     int mError;
     QTimer mCommitQueueTimer;
diff --git a/common/sourcewriteback.cpp b/common/sourcewriteback.cpp
deleted file mode 100644
index e2994d1..0000000
--- a/common/sourcewriteback.cpp
+++ /dev/null
@@ -1,146 +0,0 @@
-/*
- * Copyright (C) 2016 Christian Mollekopf <mollekopf@kolabsys.com>
- *
- * This library is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Lesser General Public
- * License as published by the Free Software Foundation; either
- * version 2.1 of the License, or (at your option) version 3, or any
- * later version accepted by the membership of KDE e.V. (or its
- * successor approved by the membership of KDE e.V.), which shall
- * act as a proxy defined in Section 6 of version 3 of the license.
- *
- * This library is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
- * Lesser General Public License for more details.
- *
- * You should have received a copy of the GNU Lesser General Public
- * License along with this library.  If not, see <http://www.gnu.org/licenses/>.
- */
-#include "sourcewriteback.h"
-
-#include "definitions.h"
-#include "log.h"
-#include "bufferutils.h"
-#include "entitybuffer.h"
-#include "entity_generated.h"
-
-#define ENTITY_TYPE_MAIL "mail"
-#define ENTITY_TYPE_FOLDER "folder"
-
-SINK_DEBUG_AREA("sourcewriteback")
-
-using namespace Sink;
-
-SourceWriteBack::SourceWriteBack(const ResourceContext &context)
-    : ChangeReplay(context),
-    mResourceContext(context),
-    mSyncStorage(Sink::storageLocation(), context.instanceId() + ".synchronization", Sink::Storage::DataStore::ReadWrite),
-    mEntityStore(QSharedPointer<Storage::EntityStore>::create(mResourceContext))
-{
-
-}
-
-Storage::EntityStore &SourceWriteBack::store()
-{
-    return *mEntityStore;
-}
-
-RemoteIdMap &SourceWriteBack::syncStore()
-{
-    if (!mSyncStore) {
-        mSyncStore = QSharedPointer<RemoteIdMap>::create(mSyncTransaction);
-    }
-    return *mSyncStore;
-}
-
-bool SourceWriteBack::canReplay(const QByteArray &type, const QByteArray &key, const QByteArray &value)
-{
-    Sink::EntityBuffer buffer(value);
-    const Sink::Entity &entity = buffer.entity();
-    const auto metadataBuffer = Sink::EntityBuffer::readBuffer<Sink::Metadata>(entity.metadata());
-    Q_ASSERT(metadataBuffer);
-    if (!metadataBuffer->replayToSource()) {
-        SinkTrace() << "Change is coming from the source";
-    }
-    return metadataBuffer->replayToSource();
-}
-
-KAsync::Job<void> SourceWriteBack::replay(const QByteArray &type, const QByteArray &key, const QByteArray &value)
-{
-    SinkTrace() << "Replaying" << type << key;
-
-    Sink::EntityBuffer buffer(value);
-    const Sink::Entity &entity = buffer.entity();
-    const auto metadataBuffer = Sink::EntityBuffer::readBuffer<Sink::Metadata>(entity.metadata());
-    Q_ASSERT(metadataBuffer);
-    Q_ASSERT(!mSyncStore);
-    Q_ASSERT(!mSyncTransaction);
-    mEntityStore->startTransaction(Storage::DataStore::ReadOnly);
-    mSyncTransaction = mSyncStorage.createTransaction(Sink::Storage::DataStore::ReadWrite);
-
-    // const qint64 revision = metadataBuffer ? metadataBuffer->revision() : -1;
-    const auto operation = metadataBuffer ? metadataBuffer->operation() : Sink::Operation_Creation;
-    const auto uid = Sink::Storage::DataStore::uidFromKey(key);
-    const auto modifiedProperties = metadataBuffer->modifiedProperties() ? BufferUtils::fromVector(*metadataBuffer->modifiedProperties()) : QByteArrayList();
-    QByteArray oldRemoteId;
-
-    if (operation != Sink::Operation_Creation) {
-        oldRemoteId = syncStore().resolveLocalId(type, uid);
-        if (oldRemoteId.isEmpty()) {
-            SinkWarning() << "Couldn't find the remote id for: " << type << uid;
-            return KAsync::error<void>(1, "Couldn't find the remote id.");
-        }
-    }
-    SinkTrace() << "Replaying " << key << type << uid << oldRemoteId;
-
-    KAsync::Job<QByteArray> job = KAsync::null<QByteArray>();
-    if (type == ENTITY_TYPE_FOLDER) {
-        auto folder = store().readEntity<ApplicationDomain::Folder>(key);
-        job = replay(folder, operation, oldRemoteId, modifiedProperties);
-    } else if (type == ENTITY_TYPE_MAIL) {
-        auto mail = store().readEntity<ApplicationDomain::Mail>(key);
-        job = replay(mail, operation, oldRemoteId, modifiedProperties);
-    }
-
-    return job.syncThen<void, QByteArray>([this, operation, type, uid, oldRemoteId](const QByteArray &remoteId) {
-        if (operation == Sink::Operation_Creation) {
-            SinkTrace() << "Replayed creation with remote id: " << remoteId;
-            if (remoteId.isEmpty()) {
-                SinkWarning() << "Returned an empty remoteId from the creation";
-            } else {
-                syncStore().recordRemoteId(type, uid, remoteId);
-            }
-        } else if (operation == Sink::Operation_Modification) {
-            SinkTrace() << "Replayed modification with remote id: " << remoteId;
-            if (remoteId.isEmpty()) {
-                SinkWarning() << "Returned an empty remoteId from the creation";
-            } else {
-               syncStore().updateRemoteId(type, uid, remoteId);
-            }
-        } else if (operation == Sink::Operation_Removal) {
-            SinkTrace() << "Replayed removal with remote id: " << oldRemoteId;
-            syncStore().removeRemoteId(type, uid, oldRemoteId);
-        } else {
-            SinkError() << "Unkown operation" << operation;
-        }
-    })
-    .syncThen<void>([this](const KAsync::Error &error) {
-        if (error) {
-            SinkWarning() << "Failed to replay change: " << error.errorMessage;
-        }
-        mSyncStore.clear();
-        mSyncTransaction.commit();
-        mEntityStore->abortTransaction();
-    });
-}
-
-KAsync::Job<QByteArray> SourceWriteBack::replay(const ApplicationDomain::Mail &, Sink::Operation, const QByteArray &, const QList<QByteArray> &)
-{
-    return KAsync::null<QByteArray>();
-}
-
-KAsync::Job<QByteArray> SourceWriteBack::replay(const ApplicationDomain::Folder &, Sink::Operation, const QByteArray &, const QList<QByteArray> &)
-{
-    return KAsync::null<QByteArray>();
-}
diff --git a/common/sourcewriteback.h b/common/sourcewriteback.h
deleted file mode 100644
index cf393e4..0000000
--- a/common/sourcewriteback.h
+++ /dev/null
@@ -1,71 +0,0 @@
-/*
- * Copyright (C) 2016 Christian Mollekopf <mollekopf@kolabsys.com>
- *
- * This library is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Lesser General Public
- * License as published by the Free Software Foundation; either
- * version 2.1 of the License, or (at your option) version 3, or any
- * later version accepted by the membership of KDE e.V. (or its
- * successor approved by the membership of KDE e.V.), which shall
- * act as a proxy defined in Section 6 of version 3 of the license.
- *
- * This library is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
- * Lesser General Public License for more details.
- *
- * You should have received a copy of the GNU Lesser General Public
- * License along with this library.  If not, see <http://www.gnu.org/licenses/>.
- */
-#pragma once
-
-#include "sink_export.h"
-
-#include "changereplay.h"
-#include "storage.h"
-#include "storage/entitystore.h"
-#include "remoteidmap.h"
-#include "metadata_generated.h"
-
-namespace Sink {
-
-/**
- * Replay changes to the source
- */
-class SINK_EXPORT SourceWriteBack : public ChangeReplay
-{
-public:
-    SourceWriteBack(const ResourceContext &resourceContext);
-
-protected:
-    ///Base implementation calls the replay$Type calls
-    virtual KAsync::Job<void> replay(const QByteArray &type, const QByteArray &key, const QByteArray &value) Q_DECL_OVERRIDE;
-    virtual bool canReplay(const QByteArray &type, const QByteArray &key, const QByteArray &value) Q_DECL_OVERRIDE;
-
-protected:
-    ///Implement to write back changes to the server
-    virtual KAsync::Job<QByteArray> replay(const Sink::ApplicationDomain::Mail &, Sink::Operation, const QByteArray &oldRemoteId, const QList<QByteArray> &);
-    virtual KAsync::Job<QByteArray> replay(const Sink::ApplicationDomain::Folder &, Sink::Operation, const QByteArray &oldRemoteId, const QList<QByteArray> &);
-
-    //Read/Write access to sync storage
-    RemoteIdMap &syncStore();
-
-    template <typename T>
-    T getPrevious(const T &entity)
-    {
-        return store().readPrevious<T>(entity.identifier(), entity.revision());
-    }
-
-private:
-    //Read only access to main storage
-    Storage::EntityStore &store();
-    ResourceContext mResourceContext;
-    Sink::Storage::DataStore mSyncStorage;
-    QSharedPointer<RemoteIdMap> mSyncStore;
-    QSharedPointer<Storage::EntityStore> mEntityStore;
-    Sink::Storage::DataStore::Transaction mSyncTransaction;
-    QByteArray mResourceType;
-    QByteArray mResourceInstanceIdentifier;
-};
-
-}
diff --git a/common/synchronizer.cpp b/common/synchronizer.cpp
index 713387e..10acefc 100644
--- a/common/synchronizer.cpp
+++ b/common/synchronizer.cpp
@@ -33,7 +33,8 @@ SINK_DEBUG_AREA("synchronizer")
 using namespace Sink;
 
 Synchronizer::Synchronizer(const Sink::ResourceContext &context)
-    : mResourceContext(context),
+    : ChangeReplay(context),
+    mResourceContext(context),
     mEntityStore(Storage::EntityStore::Ptr::create(mResourceContext)),
     mSyncStorage(Sink::storageLocation(), mResourceContext.instanceId() + ".synchronization", Sink::Storage::DataStore::DataStore::ReadWrite)
 {
@@ -310,6 +311,97 @@ Sink::Storage::DataStore::DataStore::Transaction &Synchronizer::syncTransaction(
     return mSyncTransaction;
 }
 
+bool Synchronizer::canReplay(const QByteArray &type, const QByteArray &key, const QByteArray &value)
+{
+    Sink::EntityBuffer buffer(value);
+    const Sink::Entity &entity = buffer.entity();
+    const auto metadataBuffer = Sink::EntityBuffer::readBuffer<Sink::Metadata>(entity.metadata());
+    Q_ASSERT(metadataBuffer);
+    if (!metadataBuffer->replayToSource()) {
+        SinkTrace() << "Change is coming from the source";
+    }
+    return metadataBuffer->replayToSource();
+}
+
+KAsync::Job<void> Synchronizer::replay(const QByteArray &type, const QByteArray &key, const QByteArray &value)
+{
+    SinkTrace() << "Replaying" << type << key;
+
+    Sink::EntityBuffer buffer(value);
+    const Sink::Entity &entity = buffer.entity();
+    const auto metadataBuffer = Sink::EntityBuffer::readBuffer<Sink::Metadata>(entity.metadata());
+    Q_ASSERT(metadataBuffer);
+    Q_ASSERT(!mSyncStore);
+    Q_ASSERT(!mSyncTransaction);
+    mEntityStore->startTransaction(Storage::DataStore::ReadOnly);
+    mSyncTransaction = mSyncStorage.createTransaction(Sink::Storage::DataStore::ReadWrite);
+
+    const auto operation = metadataBuffer ? metadataBuffer->operation() : Sink::Operation_Creation;
+    const auto uid = Sink::Storage::DataStore::uidFromKey(key);
+    const auto modifiedProperties = metadataBuffer->modifiedProperties() ? BufferUtils::fromVector(*metadataBuffer->modifiedProperties()) : QByteArrayList();
+    QByteArray oldRemoteId;
+
+    if (operation != Sink::Operation_Creation) {
+        oldRemoteId = syncStore().resolveLocalId(type, uid);
+        if (oldRemoteId.isEmpty()) {
+            SinkWarning() << "Couldn't find the remote id for: " << type << uid;
+            return KAsync::error<void>(1, "Couldn't find the remote id.");
+        }
+    }
+    SinkTrace() << "Replaying " << key << type << uid << oldRemoteId;
+
+    KAsync::Job<QByteArray> job = KAsync::null<QByteArray>();
+    //TODO This requires supporting every domain type here as well. Can we solve this better so we can do the dispatch somewhere centrally?
+    if (type == ApplicationDomain::getTypeName<ApplicationDomain::Folder>()) {
+        auto folder = store().readEntity<ApplicationDomain::Folder>(key);
+        job = replay(folder, operation, oldRemoteId, modifiedProperties);
+    } else if (type == ApplicationDomain::getTypeName<ApplicationDomain::Mail>()) {
+        auto mail = store().readEntity<ApplicationDomain::Mail>(key);
+        job = replay(mail, operation, oldRemoteId, modifiedProperties);
+    }
+
+    return job.syncThen<void, QByteArray>([this, operation, type, uid, oldRemoteId](const QByteArray &remoteId) {
+        if (operation == Sink::Operation_Creation) {
+            SinkTrace() << "Replayed creation with remote id: " << remoteId;
+            if (remoteId.isEmpty()) {
+                SinkWarning() << "Returned an empty remoteId from the creation";
+            } else {
+                syncStore().recordRemoteId(type, uid, remoteId);
+            }
+        } else if (operation == Sink::Operation_Modification) {
+            SinkTrace() << "Replayed modification with remote id: " << remoteId;
+            if (remoteId.isEmpty()) {
+                SinkWarning() << "Returned an empty remoteId from the creation";
+            } else {
+               syncStore().updateRemoteId(type, uid, remoteId);
+            }
+        } else if (operation == Sink::Operation_Removal) {
+            SinkTrace() << "Replayed removal with remote id: " << oldRemoteId;
+            syncStore().removeRemoteId(type, uid, oldRemoteId);
+        } else {
+            SinkError() << "Unkown operation" << operation;
+        }
+    })
+    .syncThen<void>([this](const KAsync::Error &error) {
+        if (error) {
+            SinkWarning() << "Failed to replay change: " << error.errorMessage;
+        }
+        mSyncStore.clear();
+        mSyncTransaction.commit();
+        mEntityStore->abortTransaction();
+    });
+}
+
+KAsync::Job<QByteArray> Synchronizer::replay(const ApplicationDomain::Mail &, Sink::Operation, const QByteArray &, const QList<QByteArray> &)
+{
+    return KAsync::null<QByteArray>();
+}
+
+KAsync::Job<QByteArray> Synchronizer::replay(const ApplicationDomain::Folder &, Sink::Operation, const QByteArray &, const QList<QByteArray> &)
+{
+    return KAsync::null<QByteArray>();
+}
+
 #define REGISTER_TYPE(T)                                                          \
     template void Synchronizer::createOrModify(const QByteArray &bufferType, const QByteArray &remoteId, const T &entity, const QHash<QByteArray, Sink::Query::Comparator> &mergeCriteria); \
     template void Synchronizer::modify(const T &entity);
diff --git a/common/synchronizer.h b/common/synchronizer.h
index 47518ee..0a51f54 100644
--- a/common/synchronizer.h
+++ b/common/synchronizer.h
@@ -27,6 +27,8 @@
 #include <messagequeue.h>
 #include <storage.h>
 #include <storage/entitystore.h>
+#include "changereplay.h"
+#include "remoteidmap.h"
 
 namespace Sink {
 class RemoteIdMap;
@@ -34,8 +36,9 @@ class RemoteIdMap;
 /**
  * Synchronize and add what we don't already have to local queue
  */
-class SINK_EXPORT Synchronizer
+class SINK_EXPORT Synchronizer : public ChangeReplay
 {
+    Q_OBJECT
 public:
     Synchronizer(const Sink::ResourceContext &resourceContext);
     virtual ~Synchronizer();
@@ -53,6 +56,16 @@ public:
     Sink::Storage::DataStore::Transaction &syncTransaction();
 
 protected:
+    ///Base implementation calls the replay$Type calls
+    virtual KAsync::Job<void> replay(const QByteArray &type, const QByteArray &key, const QByteArray &value) Q_DECL_OVERRIDE;
+    virtual bool canReplay(const QByteArray &type, const QByteArray &key, const QByteArray &value) Q_DECL_OVERRIDE;
+
+protected:
+    ///Implement to write back changes to the server
+    virtual KAsync::Job<QByteArray> replay(const Sink::ApplicationDomain::Mail &, Sink::Operation, const QByteArray &oldRemoteId, const QList<QByteArray> &);
+    virtual KAsync::Job<QByteArray> replay(const Sink::ApplicationDomain::Folder &, Sink::Operation, const QByteArray &oldRemoteId, const QList<QByteArray> &);
+
+protected:
     ///Calls the callback to enqueue the command
     void enqueueCommand(int commandId, const QByteArray &data);