Added the flush command.

Instead of trying to actually flush queues, we send a special command
through the same queues as the other commands and can thus guarantee
that the respective commands have been processed without blocking
anything.

1 files changed, 7 insertions, 12 deletions

diff --git a/common/listener.cpp b/common/listener.cpp
index c3c6bc2..2ab0333 100644
--- a/common/listener.cpp
+++ b/common/listener.cpp
@@ -33,7 +33,6 @@
 #include "common/synchronize_generated.h"
 #include "common/notification_generated.h"
 #include "common/revisionreplayed_generated.h"
-#include "common/inspection_generated.h"
 
 #include <QLocalServer>
 #include <QLocalSocket>
@@ -244,18 +243,13 @@ void Listener::processCommand(int commandId, uint messageId, const QByteArray &c
                 auto timer = QSharedPointer<QTime>::create();
                 timer->start();
                 auto job = KAsync::null<void>();
-                if (buffer->sourceSync()) {
-                    Sink::QueryBase query;
-                    if (buffer->query()) {
-                        auto data = QByteArray::fromStdString(buffer->query()->str());
-                        QDataStream stream(&data, QIODevice::ReadOnly);
-                        stream >> query;
-                    }
-                    job = loadResource().synchronizeWithSource(query);
-                }
-                if (buffer->localSync()) {
-                    job = job.then<void>(loadResource().processAllMessages());
+                Sink::QueryBase query;
+                if (buffer->query()) {
+                    auto data = QByteArray::fromStdString(buffer->query()->str());
+                    QDataStream stream(&data, QIODevice::ReadOnly);
+                    stream >> query;
                 }
+                job = loadResource().synchronizeWithSource(query);
                 job.then<void>([callback, timer](const KAsync::Error &error) {
                         if (error) {
                             SinkWarning() << "Sync failed: " << error.errorMessage;
@@ -279,6 +273,7 @@ void Listener::processCommand(int commandId, uint messageId, const QByteArray &c
         case Sink::Commands::DeleteEntityCommand:
         case Sink::Commands::ModifyEntityCommand:
         case Sink::Commands::CreateEntityCommand:
+        case Sink::Commands::FlushCommand:
             SinkTrace() << "Command id  " << messageId << " of type \"" << Sink::Commands::name(commandId) << "\" from " << client.name;
             loadResource().processCommand(commandId, commandBuffer);
             break;