1 files changed, 127 insertions, 41 deletions

diff --git a/common/synchronizer.cpp b/common/synchronizer.cpp
index b147615..3e7bd30 100644
--- a/common/synchronizer.cpp
+++ b/common/synchronizer.cpp
@@ -33,13 +33,14 @@
 using namespace Sink;
 
 Synchronizer::Synchronizer(const Sink::ResourceContext &context)
-    : ChangeReplay(context),
+    : ChangeReplay(context, {"synchronizer"}),
     mLogCtx{"synchronizer"},
     mResourceContext(context),
     mEntityStore(Storage::EntityStore::Ptr::create(mResourceContext, mLogCtx)),
     mSyncStorage(Sink::storageLocation(), mResourceContext.instanceId() + ".synchronization", Sink::Storage::DataStore::DataStore::ReadWrite),
     mSyncInProgress(false)
 {
+    mCurrentState.push(ApplicationDomain::Status::OfflineStatus);
     SinkTraceCtx(mLogCtx) << "Starting synchronizer: " << mResourceContext.resourceType << mResourceContext.instanceId();
 }
 
@@ -252,15 +253,21 @@ void Synchronizer::modify(const DomainType &entity, const QByteArray &newResourc
 
 QList<Synchronizer::SyncRequest> Synchronizer::getSyncRequests(const Sink::QueryBase &query)
 {
-    QList<Synchronizer::SyncRequest> list;
-    list << Synchronizer::SyncRequest{query, "sync"};
-    return list;
+    return QList<Synchronizer::SyncRequest>() << Synchronizer::SyncRequest{query, "sync"};
+}
+
+void Synchronizer::mergeIntoQueue(const Synchronizer::SyncRequest &request, QList<Synchronizer::SyncRequest> &queue)
+{
+    mSyncRequestQueue << request;
 }
 
 void Synchronizer::synchronize(const Sink::QueryBase &query)
 {
     SinkTraceCtx(mLogCtx) << "Synchronizing";
-    mSyncRequestQueue << getSyncRequests(query);
+    auto newRequests = getSyncRequests(query);
+    for (const auto &request: newRequests) {
+        mergeIntoQueue(request, mSyncRequestQueue);
+    }
     processSyncQueue().exec();
 }
 
@@ -286,6 +293,42 @@ void Synchronizer::flushComplete(const QByteArray &flushId)
     }
 }
 
+void Synchronizer::emitNotification(Notification::NoticationType type, int code, const QString &message, const QByteArray &id, const QByteArrayList &entities)
+{
+    Sink::Notification n;
+    n.id = id;
+    n.type = type;
+    n.message = message;
+    n.code = code;
+    n.entities = entities;
+    emit notify(n);
+}
+
+void Synchronizer::reportProgress(int progress, int total)
+{
+    SinkLogCtx(mLogCtx) << "Progress: " << progress << " out of " << total;
+}
+
+void Synchronizer::setStatusFromResult(const KAsync::Error &error, const QString &s, const QByteArray &requestId)
+{
+    if (error) {
+        if (error.errorCode == ApplicationDomain::ConnectionError) {
+            //Couldn't connect, so we assume we don't have a network connection.
+            setStatus(ApplicationDomain::OfflineStatus, s, requestId);
+        } else if (error.errorCode == ApplicationDomain::ConfigurationError) {
+            //There is an error with the configuration.
+            setStatus(ApplicationDomain::ErrorStatus, s, requestId);
+        } else if (error.errorCode == ApplicationDomain::LoginError) {
+            //If we failed to login altough we could connect that indicates a problem with our setup.
+            setStatus(ApplicationDomain::ErrorStatus, s, requestId);
+        }
+        //We don't know what kind of error this was, so we assume it's transient and don't change ou status.
+    } else {
+        //An operation against the server worked, so we're probably online.
+        setStatus(ApplicationDomain::ConnectedStatus, s, requestId);
+    }
+}
+
 KAsync::Job<void> Synchronizer::processRequest(const SyncRequest &request)
 {
     if (request.options & SyncRequest::RequestFlush) {
@@ -310,35 +353,21 @@ KAsync::Job<void> Synchronizer::processRequest(const SyncRequest &request)
         });
     } else if (request.requestType == Synchronizer::SyncRequest::Synchronization) {
         return KAsync::start([this, request] {
-            Sink::Notification n;
-            n.id = request.requestId;
-            n.type = Notification::Status;
-            n.message = "Synchronization has started.";
-            n.code = ApplicationDomain::BusyStatus;
-            emit notify(n);
             SinkLogCtx(mLogCtx) << "Synchronizing: " << request.query;
+            emitNotification(Notification::Info, ApplicationDomain::SyncInProgress, {}, {}, request.applicableEntities);
         }).then(synchronizeWithSource(request.query)).then([this] {
             //Commit after every request, so implementations only have to commit more if they add a lot of data.
             commit();
         }).then<void>([this, request](const KAsync::Error &error) {
+            setStatusFromResult(error, "Synchronization has ended.", request.requestId);
             if (error) {
                 //Emit notification with error
-                SinkWarningCtx(mLogCtx) << "Synchronization failed: " << error.errorMessage;
-                Sink::Notification n;
-                n.id = request.requestId;
-                n.type = Notification::Status;
-                n.message = "Synchronization has ended.";
-                n.code = ApplicationDomain::ErrorStatus;
-                emit notify(n);
+                SinkWarningCtx(mLogCtx) << "Synchronization failed: " << error;
+                emitNotification(Notification::Warning, ApplicationDomain::SyncError, {}, {}, request.applicableEntities);
                 return KAsync::error(error);
             } else {
                 SinkLogCtx(mLogCtx) << "Done Synchronizing";
-                Sink::Notification n;
-                n.id = request.requestId;
-                n.type = Notification::Status;
-                n.message = "Synchronization has ended.";
-                n.code = ApplicationDomain::ConnectedStatus;
-                emit notify(n);
+                emitNotification(Notification::Info, ApplicationDomain::SyncSuccess, {}, {}, request.applicableEntities);
                 return KAsync::null();
             }
         });
@@ -347,11 +376,8 @@ KAsync::Job<void> Synchronizer::processRequest(const SyncRequest &request)
             Q_ASSERT(!request.requestId.isEmpty());
             //FIXME it looks like this is emitted before the replay actually finishes
             if (request.flushType == Flush::FlushReplayQueue) {
-                SinkTraceCtx(mLogCtx) << "Emitting flush completion.";
-                Sink::Notification n;
-                n.type = Sink::Notification::FlushCompletion;
-                n.id = request.requestId;
-                emit notify(n);
+                SinkTraceCtx(mLogCtx) << "Emitting flush completion: " << request.requestId;
+                emitNotification(Notification::FlushCompletion, 0, "", request.requestId);
             } else {
                 flatbuffers::FlatBufferBuilder fbb;
                 auto flushId = fbb.CreateString(request.requestId.toStdString());
@@ -361,7 +387,24 @@ KAsync::Job<void> Synchronizer::processRequest(const SyncRequest &request)
             }
         });
     } else if (request.requestType == Synchronizer::SyncRequest::ChangeReplay) {
-        return replayNextRevision();
+        if (ChangeReplay::allChangesReplayed()) {
+            return KAsync::null();
+        } else {
+            return KAsync::start([this, request] {
+                SinkLogCtx(mLogCtx) << "Replaying changes.";
+            })
+            .then(replayNextRevision())
+            .then<void>([this, request](const KAsync::Error &error) {
+                setStatusFromResult(error, "Changereplay has ended.", "changereplay");
+                if (error) {
+                    SinkWarningCtx(mLogCtx) << "Changereplay failed: " << error.errorMessage;
+                    return KAsync::error(error);
+                } else {
+                    SinkLogCtx(mLogCtx) << "Done replaying changes";
+                    return KAsync::null();
+                }
+            });
+        }
     } else {
         SinkWarningCtx(mLogCtx) << "Unknown request type: " << request.requestType;
         return KAsync::error(KAsync::Error{"Unknown request type."});
@@ -369,6 +412,34 @@ KAsync::Job<void> Synchronizer::processRequest(const SyncRequest &request)
 
 }
 
+void Synchronizer::setStatus(ApplicationDomain::Status state, const QString &reason, const QByteArray requestId)
+{
+    if (state != mCurrentState.top()) {
+        if (mCurrentState.top() == ApplicationDomain::BusyStatus) {
+            mCurrentState.pop();
+        }
+        mCurrentState.push(state);
+        emitNotification(Notification::Status, state, reason, requestId);
+    }
+}
+
+void Synchronizer::resetStatus(const QByteArray requestId)
+{
+    mCurrentState.pop();
+    emitNotification(Notification::Status, mCurrentState.top(), {}, requestId);
+}
+
+void Synchronizer::setBusy(bool busy, const QString &reason, const QByteArray requestId)
+{
+    if (busy) {
+        setStatus(ApplicationDomain::BusyStatus, reason, requestId);
+    } else {
+        if (mCurrentState.top() == ApplicationDomain::BusyStatus) {
+            resetStatus(requestId);
+        }
+    }
+}
+
 KAsync::Job<void> Synchronizer::processSyncQueue()
 {
     if (mSyncRequestQueue.isEmpty()) {
@@ -387,14 +458,20 @@ KAsync::Job<void> Synchronizer::processSyncQueue()
     }
 
     const auto request = mSyncRequestQueue.takeFirst();
-    return KAsync::start([this] {
+    return KAsync::start([=] {
         mMessageQueue->startTransaction();
         mEntityStore->startTransaction(Sink::Storage::DataStore::ReadOnly);
         mSyncInProgress = true;
+        if (request.requestType == Synchronizer::SyncRequest::Synchronization) {
+            setBusy(true, "Synchronization has started.", request.requestId);
+        } else if (request.requestType == Synchronizer::SyncRequest::ChangeReplay) {
+            setBusy(true, "ChangeReplay has started.", "changereplay");
+        }
     })
     .then(processRequest(request))
-    .then<void>([this](const KAsync::Error &error) {
+    .then<void>([this, request](const KAsync::Error &error) {
         SinkTraceCtx(mLogCtx) << "Sync request processed";
+        setBusy(false, {}, request.requestId);
         mEntityStore->abortTransaction();
         mSyncTransaction.abort();
         mMessageQueue->commit();
@@ -404,8 +481,8 @@ KAsync::Job<void> Synchronizer::processSyncQueue()
             emit changesReplayed();
         }
         if (error) {
-            SinkWarningCtx(mLogCtx) << "Error during sync: " << error.errorMessage;
-            return KAsync::error(error);
+            SinkWarningCtx(mLogCtx) << "Error during sync: " << error;
+            emitNotification(Notification::Error, error.errorCode, error.errorMessage, request.requestId);
         }
         //In case we got more requests meanwhile.
         return processSyncQueue();
@@ -499,6 +576,12 @@ KAsync::Job<void> Synchronizer::replay(const QByteArray &type, const QByteArray
     } else if (type == ApplicationDomain::getTypeName<ApplicationDomain::Mail>()) {
         auto mail = store().readEntity<ApplicationDomain::Mail>(key);
         job = replay(mail, operation, oldRemoteId, modifiedProperties);
+    } else if (type == ApplicationDomain::getTypeName<ApplicationDomain::Contact>()) {
+        auto mail = store().readEntity<ApplicationDomain::Contact>(key);
+        job = replay(mail, operation, oldRemoteId, modifiedProperties);
+    } else if (type == ApplicationDomain::getTypeName<ApplicationDomain::Addressbook>()) {
+        auto mail = store().readEntity<ApplicationDomain::Addressbook>(key);
+        job = replay(mail, operation, oldRemoteId, modifiedProperties);
     } else {
         SinkErrorCtx(mLogCtx) << "Replayed unknown type: " << type;
     }
@@ -506,21 +589,19 @@ KAsync::Job<void> Synchronizer::replay(const QByteArray &type, const QByteArray
     return job.then([this, operation, type, uid, oldRemoteId](const QByteArray &remoteId) {
         if (operation == Sink::Operation_Creation) {
             SinkTraceCtx(mLogCtx) << "Replayed creation with remote id: " << remoteId;
-            if (remoteId.isEmpty()) {
-                SinkWarningCtx(mLogCtx) << "Returned an empty remoteId from the creation";
-            } else {
+            if (!remoteId.isEmpty()) {
                 syncStore().recordRemoteId(type, uid, remoteId);
             }
         } else if (operation == Sink::Operation_Modification) {
             SinkTraceCtx(mLogCtx) << "Replayed modification with remote id: " << remoteId;
-            if (remoteId.isEmpty()) {
-                SinkWarningCtx(mLogCtx) << "Returned an empty remoteId from the modification";
-            } else {
+            if (!remoteId.isEmpty()) {
                syncStore().updateRemoteId(type, uid, remoteId);
             }
         } else if (operation == Sink::Operation_Removal) {
             SinkTraceCtx(mLogCtx) << "Replayed removal with remote id: " << oldRemoteId;
-            syncStore().removeRemoteId(type, uid, oldRemoteId);
+            if (!oldRemoteId.isEmpty()) {
+                syncStore().removeRemoteId(type, uid, oldRemoteId);
+            }
         } else {
             SinkErrorCtx(mLogCtx) << "Unkown operation" << operation;
         }
@@ -539,6 +620,11 @@ KAsync::Job<QByteArray> Synchronizer::replay(const ApplicationDomain::Contact &,
     return KAsync::null<QByteArray>();
 }
 
+KAsync::Job<QByteArray> Synchronizer::replay(const ApplicationDomain::Addressbook &, Sink::Operation, const QByteArray &, const QList<QByteArray> &)
+{
+    return KAsync::null<QByteArray>();
+}
+
 KAsync::Job<QByteArray> Synchronizer::replay(const ApplicationDomain::Mail &, Sink::Operation, const QByteArray &, const QList<QByteArray> &)
 {
     return KAsync::null<QByteArray>();