Added progress notification

6 files changed, 34 insertions, 2 deletions

diff --git a/common/commands/notification.fbs b/common/commands/notification.fbs
index 517111c..7ced666 100644
--- a/common/commands/notification.fbs
+++ b/common/commands/notification.fbs
@@ -5,6 +5,8 @@ table Notification {
     identifier: string; //An identifier that links back to the something related to the notification (e.g. a command id)
     message: string;
     code: int = 0; //See notification.h
+    progress: int = 0; //See notification.h
+    total: int = 0; //See notification.h
     entities: [string]; //A list of entities this applies to
 }
 
diff --git a/common/listener.cpp b/common/listener.cpp
index 983e438..ec2bedb 100644
--- a/common/listener.cpp
+++ b/common/listener.cpp
@@ -420,6 +420,8 @@ void Listener::notify(const Sink::Notification &notification)
     builder.add_identifier(idString);
     builder.add_message(messageString);
     builder.add_entities(entities);
+    builder.add_progress(notification.progress);
+    builder.add_total(notification.total);
     auto command = builder.Finish();
     Sink::Commands::FinishNotificationBuffer(m_fbb, command);
     for (Client &client : m_connections) {
diff --git a/common/notification.h b/common/notification.h
index f5379fd..30e240c 100644
--- a/common/notification.h
+++ b/common/notification.h
@@ -56,6 +56,8 @@ public:
     QString message;
     //A return code. Zero typically indicates success.
     int code = 0;
+    int progress = 0;
+    int total = 0;
     QByteArray resource;
 };
 }
diff --git a/common/resourceaccess.cpp b/common/resourceaccess.cpp
index ad8cae9..cf8b2e0 100644
--- a/common/resourceaccess.cpp
+++ b/common/resourceaccess.cpp
@@ -547,6 +547,8 @@ static Sink::Notification getNotification(const Sink::Commands::Notification *bu
     }
     n.type = buffer->type();
     n.code = buffer->code();
+    n.progress = buffer->progress();
+    n.total = buffer->total();
     n.entities = BufferUtils::fromVector(*buffer->entities());
     return n;
 }
diff --git a/common/synchronizer.cpp b/common/synchronizer.cpp
index 3e7bd30..58e5e7a 100644
--- a/common/synchronizer.cpp
+++ b/common/synchronizer.cpp
@@ -304,9 +304,21 @@ void Synchronizer::emitNotification(Notification::NoticationType type, int code,
     emit notify(n);
 }
 
+void Synchronizer::emitProgressNotification(Notification::NoticationType type, int progress, int total, const QByteArray &id, const QByteArrayList &entities)
+{
+    Sink::Notification n;
+    n.id = id;
+    n.type = type;
+    n.progress = progress;
+    n.total = total;
+    n.entities = entities;
+    emit notify(n);
+}
+
 void Synchronizer::reportProgress(int progress, int total)
 {
     SinkLogCtx(mLogCtx) << "Progress: " << progress << " out of " << total;
+    emitProgressNotification(Notification::Progress, progress, total, mCurrentRequest.requestId, mCurrentRequest.applicableEntities);
 }
 
 void Synchronizer::setStatusFromResult(const KAsync::Error &error, const QString &s, const QByteArray &requestId)
@@ -465,13 +477,15 @@ KAsync::Job<void> Synchronizer::processSyncQueue()
         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");
+            setBusy(true, "ChangeReplay has started.", request.requestId);
         }
+        mCurrentRequest = request;
     })
     .then(processRequest(request))
     .then<void>([this, request](const KAsync::Error &error) {
         SinkTraceCtx(mLogCtx) << "Sync request processed";
         setBusy(false, {}, request.requestId);
+        mCurrentRequest = {};
         mEntityStore->abortTransaction();
         mSyncTransaction.abort();
         mMessageQueue->commit();
@@ -516,7 +530,7 @@ void Synchronizer::revisionChanged()
             return;
         }
     }
-    mSyncRequestQueue << Synchronizer::SyncRequest{Synchronizer::SyncRequest::ChangeReplay};
+    mSyncRequestQueue << Synchronizer::SyncRequest{Synchronizer::SyncRequest::ChangeReplay, "changereplay"};
     processSyncQueue().exec();
 }
 
diff --git a/common/synchronizer.h b/common/synchronizer.h
index b1ee122..bb24c2b 100644
--- a/common/synchronizer.h
+++ b/common/synchronizer.h
@@ -131,6 +131,8 @@ protected:
             RequestFlush
         };
 
+        SyncRequest() = default;
+
         SyncRequest(const Sink::QueryBase &q, const QByteArray &requestId_ = QByteArray(), RequestOptions o = NoOptions)
             : requestId(requestId_),
             requestType(Synchronization),
@@ -145,6 +147,12 @@ protected:
         {
         }
 
+        SyncRequest(RequestType type, const QByteArray &requestId_)
+            : requestId(requestId_),
+            requestType(type)
+        {
+        }
+
         SyncRequest(RequestType type, int flushType_, const QByteArray &requestId_)
             : flushType(flushType_),
             requestId(requestId_),
@@ -184,6 +192,7 @@ protected:
     virtual void mergeIntoQueue(const Synchronizer::SyncRequest &request, QList<Synchronizer::SyncRequest> &queue);
 
     void emitNotification(Notification::NoticationType type, int code, const QString &message, const QByteArray &id = QByteArray{}, const QByteArrayList &entiteis = QByteArrayList{});
+    void emitProgressNotification(Notification::NoticationType type, int progress, int total, const QByteArray &id, const QByteArrayList &entities);
 
     /**
      * Report progress for current task
@@ -211,6 +220,7 @@ private:
     Sink::Storage::DataStore::Transaction mSyncTransaction;
     std::function<void(int commandId, const QByteArray &data)> mEnqueue;
     QList<SyncRequest> mSyncRequestQueue;
+    SyncRequest mCurrentRequest;
     MessageQueue *mMessageQueue;
     bool mSyncInProgress;
     QMultiHash<QByteArray, SyncRequest> mPendingSyncRequests;