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.

20 files changed, 227 insertions, 67 deletions

diff --git a/common/CMakeLists.txt b/common/CMakeLists.txt
index 5ba524b..8a16af4 100644
--- a/common/CMakeLists.txt
+++ b/common/CMakeLists.txt
@@ -95,6 +95,7 @@ generate_flatbuffers(
     commands/notification
     commands/revisionreplayed
     commands/inspection
+    commands/flush
     domain/event
     domain/mail
     domain/folder
@@ -130,6 +131,7 @@ install(FILES
     bufferadaptor.h
     test.h
     log.h
+    flush.h
     ${CMAKE_CURRENT_BINARY_DIR}/sink_export.h
     DESTINATION ${INCLUDE_INSTALL_DIR}/${PROJECT_NAME} COMPONENT Devel
 )
diff --git a/common/commands.cpp b/common/commands.cpp
index 91657b8..c0781f6 100644
--- a/common/commands.cpp
+++ b/common/commands.cpp
@@ -61,6 +61,8 @@ QByteArray name(int commandId)
             return "Inspection";
         case RemoveFromDiskCommand:
             return "RemoveFromDisk";
+        case FlushCommand:
+            return "Flush";
         case CustomCommand:
             return "Custom";
     };
diff --git a/common/commands.h b/common/commands.h
index b97bbc6..0da1b3c 100644
--- a/common/commands.h
+++ b/common/commands.h
@@ -48,6 +48,7 @@ enum CommandIds
     RevisionReplayedCommand,
     InspectionCommand,
     RemoveFromDiskCommand,
+    FlushCommand,
     CustomCommand = 0xffff
 };
 
diff --git a/common/commands/flush.fbs b/common/commands/flush.fbs
new file mode 100644
index 0000000..179f760
--- /dev/null
+++ b/common/commands/flush.fbs
@@ -0,0 +1,8 @@
+namespace Sink.Commands;
+
+table Flush {
+    id: string;
+    type: int; //See flush.h
+}
+
+root_type Flush;
diff --git a/common/commands/synchronize.fbs b/common/commands/synchronize.fbs
index 62f4b2b..7b32305 100644
--- a/common/commands/synchronize.fbs
+++ b/common/commands/synchronize.fbs
@@ -1,8 +1,6 @@
 namespace Sink.Commands;
 
 table Synchronize {
-    sourceSync: bool; //Synchronize with source
-    localSync: bool; //Ensure all queues are processed so the local state is up-to date.
     query: string;
 }
 
diff --git a/common/flush.h b/common/flush.h
new file mode 100644
index 0000000..3f04608
--- /dev/null
+++ b/common/flush.h
@@ -0,0 +1,45 @@
+/*
+ * 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"
+
+namespace Sink {
+namespace Flush {
+
+enum FlushType {
+    /**
+     * Guarantees that any commands issued before this flush are written back to the server once this flush completes.
+     * Note that this does not guarantee the success of writeback, only that an attempt has been made.
+     */
+    FlushReplayQueue,
+    /**
+     * Guarantees that any synchronization request issued before  this flush has been executed and that all entities created by it have been processed once this flush completes.
+     */
+    FlushSynchronization,
+    /**
+     * Guarantees that any modification issued before this flush has been processed once this flush completes.
+     */
+    FlushUserQueue
+};
+
+}
+}
+
diff --git a/common/genericresource.cpp b/common/genericresource.cpp
index 7c4d4ea..7b83957 100644
--- a/common/genericresource.cpp
+++ b/common/genericresource.cpp
@@ -27,6 +27,7 @@
 #include "deleteentity_generated.h"
 #include "inspection_generated.h"
 #include "notification_generated.h"
+#include "flush_generated.h"
 #include "domainadaptor.h"
 #include "commands.h"
 #include "index.h"
@@ -54,6 +55,7 @@ class CommandProcessor : public QObject
 {
     Q_OBJECT
     typedef std::function<KAsync::Job<void>(void const *, size_t)> InspectionFunction;
+    typedef std::function<KAsync::Job<void>(void const *, size_t)> FlushFunction;
     SINK_DEBUG_AREA("commandprocessor")
 
 public:
@@ -75,6 +77,11 @@ public:
         mInspect = f;
     }
 
+    void setFlushCommand(const FlushFunction &f)
+    {
+        mFlush = f;
+    }
+
 signals:
     void error(int errorCode, const QString &errorMessage);
 
@@ -124,6 +131,13 @@ private slots:
                 } else {
                     return KAsync::error<qint64>(-1, "Missing inspection command.");
                 }
+            case Sink::Commands::FlushCommand:
+                if (mFlush) {
+                    return mFlush(queuedCommand->command()->Data(), queuedCommand->command()->size())
+                        .syncThen<qint64>([]() { return -1; });
+                } else {
+                    return KAsync::error<qint64>(-1, "Missing inspection command.");
+                }
             default:
                 return KAsync::error<qint64>(-1, "Unhandled command");
         }
@@ -219,6 +233,7 @@ private:
     // The lowest revision we no longer need
     qint64 mLowerBoundRevision;
     InspectionFunction mInspect;
+    FlushFunction mFlush;
 };
 
 GenericResource::GenericResource(const ResourceContext &resourceContext, const QSharedPointer<Pipeline> &pipeline )
@@ -266,6 +281,26 @@ GenericResource::GenericResource(const ResourceContext &resourceContext, const Q
         }
         return KAsync::error<void>(-1, "Invalid inspection command.");
     });
+    mProcessor->setFlushCommand([this](void const *command, size_t size) {
+        flatbuffers::Verifier verifier((const uint8_t *)command, size);
+        if (Sink::Commands::VerifyFlushBuffer(verifier)) {
+            auto buffer = Sink::Commands::GetFlush(command);
+            const auto flushType = buffer->type();
+            const auto flushId = BufferUtils::extractBuffer(buffer->id());
+            if (flushType == Sink::Flush::FlushReplayQueue) {
+                SinkTrace() << "Flushing synchronizer ";
+                mSynchronizer->flush(flushType, flushId);
+            } else {
+                SinkTrace() << "Emitting flush completion" << flushId;
+                Sink::Notification n;
+                n.type = Sink::Notification::FlushCompletion;
+                n.id = flushId;
+                emit notify(n);
+            }
+            return KAsync::null<void>();
+        }
+        return KAsync::error<void>(-1, "Invalid flush command.");
+    });
     {
         auto ret = QObject::connect(mProcessor.get(), &CommandProcessor::error, [this](int errorCode, const QString &msg) { onProcessorError(errorCode, msg); });
         Q_ASSERT(ret);
@@ -371,6 +406,10 @@ void GenericResource::enqueueCommand(MessageQueue &mq, int commandId, const QByt
 
 void GenericResource::processCommand(int commandId, const QByteArray &data)
 {
+    if (commandId == Commands::FlushCommand) {
+        processFlushCommand(data);
+        return;
+    }
     static int modifications = 0;
     mUserQueue.startTransaction();
     enqueueCommand(mUserQueue, commandId, data);
@@ -384,6 +423,24 @@ void GenericResource::processCommand(int commandId, const QByteArray &data)
     }
 }
 
+void GenericResource::processFlushCommand(const QByteArray &data)
+{
+    flatbuffers::Verifier verifier((const uint8_t *)data.constData(), data.size());
+    if (Sink::Commands::VerifyFlushBuffer(verifier)) {
+        auto buffer = Sink::Commands::GetFlush(data.constData());
+        const auto flushType = buffer->type();
+        const auto flushId = BufferUtils::extractBuffer(buffer->id());
+        if (flushType == Sink::Flush::FlushSynchronization) {
+            mSynchronizer->flush(flushType, flushId);
+        } else {
+            mUserQueue.startTransaction();
+            enqueueCommand(mUserQueue, Commands::FlushCommand, data);
+            mUserQueue.commit();
+        }
+    }
+
+}
+
 KAsync::Job<void> GenericResource::synchronizeWithSource(const Sink::QueryBase &query)
 {
     return KAsync::start<void>([this, query] {
diff --git a/common/genericresource.h b/common/genericresource.h
index 3f92e93..9447c8b 100644
--- a/common/genericresource.h
+++ b/common/genericresource.h
@@ -47,6 +47,7 @@ public:
     virtual ~GenericResource();
 
     virtual void processCommand(int commandId, const QByteArray &data) Q_DECL_OVERRIDE;
+    virtual void processFlushCommand(const QByteArray &data);
     virtual KAsync::Job<void> synchronizeWithSource(const Sink::QueryBase &query) Q_DECL_OVERRIDE;
     virtual KAsync::Job<void> processAllMessages() Q_DECL_OVERRIDE;
     virtual void setLowerBoundRevision(qint64 revision) Q_DECL_OVERRIDE;
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;
diff --git a/common/notification.h b/common/notification.h
index dcf00a3..b1bd290 100644
--- a/common/notification.h
+++ b/common/notification.h
@@ -37,17 +37,19 @@ public:
         Warning,
         Progress,
         Inspection,
-        RevisionUpdate
+        RevisionUpdate,
+        FlushCompletion
     };
     enum InspectionCode {
-        Success,
+        Success = 0,
         Failure
     };
 
     QByteArray id;
-    int type;
+    int type = 0;
     QString message;
-    int code;
+    //A return code. Zero typically indicates success.
+    int code = 0;
 };
 }
 
diff --git a/common/query.h b/common/query.h
index 0bc5141..2adb7e9 100644
--- a/common/query.h
+++ b/common/query.h
@@ -465,6 +465,12 @@ class SyncScope : public QueryBase {
 public:
     using QueryBase::QueryBase;
 
+    SyncScope(const QueryBase &other)
+        : QueryBase(other)
+    {
+
+    }
+
     Query::Filter getResourceFilter() const
     {
         return mResourceFilter;
diff --git a/common/queuedcommand.fbs b/common/queuedcommand.fbs
index 06226d3..114e2cd 100644
--- a/common/queuedcommand.fbs
+++ b/common/queuedcommand.fbs
@@ -3,9 +3,6 @@ namespace Sink;
 table QueuedCommand {
     commandId: int;
     command: [ubyte];
-    // entityId: string;
-    // sourceRevision: ulong;
-    // targetRevision: [ubyte];
 }
 
 root_type QueuedCommand;
diff --git a/common/resourceaccess.cpp b/common/resourceaccess.cpp
index 822b5cd..b46e8b2 100644
--- a/common/resourceaccess.cpp
+++ b/common/resourceaccess.cpp
@@ -31,6 +31,7 @@
 #include "common/deleteentity_generated.h"
 #include "common/revisionreplayed_generated.h"
 #include "common/inspection_generated.h"
+#include "common/flush_generated.h"
 #include "common/entitybuffer.h"
 #include "common/bufferutils.h"
 #include "common/test.h"
@@ -291,16 +292,6 @@ KAsync::Job<void> ResourceAccess::sendCommand(int commandId, flatbuffers::FlatBu
     });
 }
 
-KAsync::Job<void> ResourceAccess::synchronizeResource(bool sourceSync, bool localSync)
-{
-    SinkTrace() << "Sending synchronize command: " << sourceSync << localSync;
-    flatbuffers::FlatBufferBuilder fbb;
-    auto command = Sink::Commands::CreateSynchronize(fbb, sourceSync, localSync);
-    Sink::Commands::FinishSynchronizeBuffer(fbb, command);
-    open();
-    return sendCommand(Commands::SynchronizeCommand, fbb);
-}
-
 KAsync::Job<void> ResourceAccess::synchronizeResource(const Sink::QueryBase &query)
 {
     flatbuffers::FlatBufferBuilder fbb;
@@ -311,8 +302,6 @@ KAsync::Job<void> ResourceAccess::synchronizeResource(const Sink::QueryBase &que
     }
     auto q = fbb.CreateString(queryString.toStdString());
     auto builder = Sink::Commands::SynchronizeBuilder(fbb);
-    builder.add_sourceSync(true);
-    builder.add_localSync(false);
     builder.add_query(q);
     Sink::Commands::FinishSynchronizeBuffer(fbb, builder.Finish());
 
@@ -390,6 +379,16 @@ ResourceAccess::sendInspectionCommand(int inspectionType, const QByteArray &insp
     return sendCommand(Sink::Commands::InspectionCommand, fbb);
 }
 
+KAsync::Job<void> ResourceAccess::sendFlushCommand(int flushType, const QByteArray &flushId)
+{
+    flatbuffers::FlatBufferBuilder fbb;
+    auto id = fbb.CreateString(flushId.toStdString());
+    auto location = Sink::Commands::CreateFlush(fbb, id, flushType);
+    Sink::Commands::FinishFlushBuffer(fbb, location);
+    open();
+    return sendCommand(Sink::Commands::FlushCommand, fbb);
+}
+
 void ResourceAccess::open()
 {
     if (d->socket && d->socket->isValid()) {
@@ -613,6 +612,8 @@ bool ResourceAccess::processMessageBuffer()
                     [[clang::fallthrough]];
                 case Sink::Notification::Warning:
                     [[clang::fallthrough]];
+                case Sink::Notification::FlushCompletion:
+                    [[clang::fallthrough]];
                 case Sink::Notification::Progress: {
                     auto n = getNotification(buffer);
                     SinkTrace() << "Received notification: " << n.type;
diff --git a/common/resourceaccess.h b/common/resourceaccess.h
index 755c8a7..4229161 100644
--- a/common/resourceaccess.h
+++ b/common/resourceaccess.h
@@ -29,6 +29,7 @@
 
 #include <flatbuffers/flatbuffers.h>
 #include "notification.h"
+#include "flush.h"
 #include "query.h"
 #include "log.h"
 
@@ -50,7 +51,6 @@ public:
     }
     virtual KAsync::Job<void> sendCommand(int commandId) = 0;
     virtual KAsync::Job<void> sendCommand(int commandId, flatbuffers::FlatBufferBuilder &fbb) = 0;
-    virtual KAsync::Job<void> synchronizeResource(bool remoteSync, bool localSync) = 0;
     virtual KAsync::Job<void> synchronizeResource(const Sink::QueryBase &filter) = 0;
 
     virtual KAsync::Job<void> sendCreateCommand(const QByteArray &uid, const QByteArray &resourceBufferType, const QByteArray &buffer)
@@ -75,6 +75,11 @@ public:
         return KAsync::null<void>();
     };
 
+    virtual KAsync::Job<void> sendFlushCommand(int flushType, const QByteArray &flushId)
+    {
+        return KAsync::null<void>();
+    }
+
     int getResourceStatus() const
     {
         return mResourceStatus;
@@ -108,7 +113,6 @@ public:
 
     KAsync::Job<void> sendCommand(int commandId) Q_DECL_OVERRIDE;
     KAsync::Job<void> sendCommand(int commandId, flatbuffers::FlatBufferBuilder &fbb) Q_DECL_OVERRIDE;
-    KAsync::Job<void> synchronizeResource(bool remoteSync, bool localSync) Q_DECL_OVERRIDE;
     KAsync::Job<void> synchronizeResource(const Sink::QueryBase &filter) Q_DECL_OVERRIDE;
     KAsync::Job<void> sendCreateCommand(const QByteArray &uid, const QByteArray &resourceBufferType, const QByteArray &buffer) Q_DECL_OVERRIDE;
     KAsync::Job<void>
@@ -117,6 +121,7 @@ public:
     KAsync::Job<void> sendRevisionReplayedCommand(qint64 revision) Q_DECL_OVERRIDE;
     KAsync::Job<void>
     sendInspectionCommand(int inspectionType,const QByteArray &inspectionId, const QByteArray &domainType, const QByteArray &entityId, const QByteArray &property, const QVariant &expecedValue) Q_DECL_OVERRIDE;
+    KAsync::Job<void> sendFlushCommand(int flushType, const QByteArray &flushId) Q_DECL_OVERRIDE;
     /**
      * Tries to connect to server, and returns a connected socket on success.
      */
diff --git a/common/resourcecontrol.cpp b/common/resourcecontrol.cpp
index 3568844..af98b8b 100644
--- a/common/resourcecontrol.cpp
+++ b/common/resourcecontrol.cpp
@@ -85,17 +85,39 @@ KAsync::Job<void> ResourceControl::flushMessageQueue(const QByteArrayList &resou
     SinkTrace() << "flushMessageQueue" << resourceIdentifier;
     return KAsync::value(resourceIdentifier)
         .template each([](const QByteArray &resource) {
-            SinkTrace() << "Flushing message queue " << resource;
-            auto resourceAccess = ResourceAccessFactory::instance().getAccess(resource, ResourceConfig::getResourceType(resource));
-            resourceAccess->open();
-            return resourceAccess->synchronizeResource(false, true)
-                .addToContext(resourceAccess);
+            return flushMessageQueue(resource);
         });
 }
 
 KAsync::Job<void> ResourceControl::flushMessageQueue(const QByteArray &resourceIdentifier)
 {
-    return flushMessageQueue(QByteArrayList() << resourceIdentifier);
+    return flush(Flush::FlushUserQueue, resourceIdentifier).then(flush(Flush::FlushSynchronization, resourceIdentifier));
+}
+
+KAsync::Job<void> ResourceControl::flush(Flush::FlushType type, const QByteArray &resourceIdentifier)
+{
+    auto resourceAccess = ResourceAccessFactory::instance().getAccess(resourceIdentifier, ResourceConfig::getResourceType(resourceIdentifier));
+    auto notifier = QSharedPointer<Sink::Notifier>::create(resourceAccess);
+    auto id = QUuid::createUuid().toByteArray();
+    return KAsync::start<void>([=](KAsync::Future<void> &future) {
+            SinkTrace() << "Waiting for notification notification " << id;
+            notifier->registerHandler([&future, id](const Notification &notification) {
+                SinkTrace() << "Received notification " << notification.type << notification.id;
+                if (notification.id == id) {
+                    SinkTrace() << "FlushComplete";
+                    if (notification.code) {
+                        SinkWarning() << "Flush return an error";
+                        future.setError(-1, "Flush returned an error: " + notification.message);
+                    } else {
+                        future.setFinished();
+                    }
+                }
+            });
+            resourceAccess->sendFlushCommand(type, id).onError([&future] (const KAsync::Error &error) {
+                SinkWarning() << "Failed to send command";
+                future.setError(1, "Failed to send command: " + error.errorMessage);
+            }).exec();
+        });
 }
 
 KAsync::Job<void> ResourceControl::flushReplayQueue(const QByteArrayList &resourceIdentifier)
diff --git a/common/resourcecontrol.h b/common/resourcecontrol.h
index 9e603e4..b910441 100644
--- a/common/resourcecontrol.h
+++ b/common/resourcecontrol.h
@@ -26,6 +26,7 @@
 #include <Async/Async>
 
 #include "inspection.h"
+#include "flush.h"
 
 namespace Sink {
 namespace ResourceControl {
@@ -58,5 +59,8 @@ KAsync::Job<void> SINK_EXPORT flushMessageQueue(const QByteArray &resourceIdenti
  */
 KAsync::Job<void> SINK_EXPORT flushReplayQueue(const QByteArrayList &resourceIdentifier);
 KAsync::Job<void> SINK_EXPORT flushReplayQueue(const QByteArray &resourceIdentifier);
+
+KAsync::Job<void> SINK_EXPORT flush(Flush::FlushType, const QByteArray &resourceIdentifier);
+
 }
 }
diff --git a/common/store.cpp b/common/store.cpp
index 6aae00f..8b8de1f 100644
--- a/common/store.cpp
+++ b/common/store.cpp
@@ -255,23 +255,7 @@ KAsync::Job<void> Store::removeDataFromDisk(const QByteArray &identifier)
 
 KAsync::Job<void> Store::synchronize(const Sink::Query &query)
 {
-    auto resources = getResources(query.getResourceFilter()).keys();
-    SinkTrace() << "synchronize" << resources;
-    return KAsync::value(resources)
-        .template each([query](const QByteArray &resource) {
-            SinkTrace() << "Synchronizing " << resource;
-            auto resourceAccess = ResourceAccessFactory::instance().getAccess(resource, ResourceConfig::getResourceType(resource));
-            return resourceAccess->synchronizeResource(true, false)
-                .addToContext(resourceAccess)
-                .then<void>([](const KAsync::Error &error) {
-                        if (error) {
-                            SinkWarning() << "Error during sync.";
-                            return KAsync::error<void>(error);
-                        }
-                        SinkTrace() << "synced.";
-                        return KAsync::null<void>();
-                    });
-        });
+    return synchronize(Sink::SyncScope{static_cast<Sink::QueryBase>(query)});
 }
 
 KAsync::Job<void> Store::synchronize(const Sink::SyncScope &scope)
diff --git a/common/synchronizer.cpp b/common/synchronizer.cpp
index 5bde597..f7dd816 100644
--- a/common/synchronizer.cpp
+++ b/common/synchronizer.cpp
@@ -27,6 +27,8 @@
 #include "createentity_generated.h"
 #include "modifyentity_generated.h"
 #include "deleteentity_generated.h"
+#include "flush_generated.h"
+#include "notification_generated.h"
 
 SINK_DEBUG_AREA("synchronizer")
 
@@ -263,6 +265,13 @@ KAsync::Job<void> Synchronizer::synchronize(const Sink::QueryBase &query)
     return processSyncQueue();
 }
 
+void Synchronizer::flush(int commandId, const QByteArray &flushId)
+{
+    SinkTrace() << "Flushing the synchronization queue";
+    mSyncRequestQueue << Synchronizer::SyncRequest{Synchronizer::SyncRequest::Flush, commandId, flushId};
+    processSyncQueue().exec();
+}
+
 KAsync::Job<void> Synchronizer::processSyncQueue()
 {
     if (mSyncRequestQueue.isEmpty() || mSyncInProgress) {
@@ -279,6 +288,20 @@ KAsync::Job<void> Synchronizer::processSyncQueue()
                 //Commit after every request, so implementations only have to commit more if they add a lot of data.
                 commit();
             });
+        } else if (request.requestType == Synchronizer::SyncRequest::Flush) {
+            if (request.flushType == Flush::FlushReplayQueue) {
+                SinkTrace() << "Emitting flush completion.";
+                Sink::Notification n;
+                n.type = Sink::Notification::FlushCompletion;
+                n.id = request.flushId;
+                emit notify(n);
+            } else {
+                flatbuffers::FlatBufferBuilder fbb;
+                auto flushId = fbb.CreateString(request.flushId);
+                auto location = Sink::Commands::CreateFlush(fbb, flushId, static_cast<int>(Sink::Flush::FlushSynchronization));
+                Sink::Commands::FinishFlushBuffer(fbb, location);
+                enqueueCommand(Sink::Commands::FlushCommand, BufferUtils::extractBuffer(fbb));
+            }
         } else {
             job = replayNextRevision();
         }
diff --git a/common/synchronizer.h b/common/synchronizer.h
index 4d5bdd5..99d4877 100644
--- a/common/synchronizer.h
+++ b/common/synchronizer.h
@@ -45,6 +45,7 @@ public:
 
     void setup(const std::function<void(int commandId, const QByteArray &data)> &enqueueCommandCallback, MessageQueue &messageQueue);
     KAsync::Job<void> synchronize(const Sink::QueryBase &query);
+    void flush(int commandId, const QByteArray &flushId);
 
     //Read only access to main storage
     Storage::EntityStore &store();
@@ -57,6 +58,9 @@ public:
 
     bool allChangesReplayed() Q_DECL_OVERRIDE;
 
+signals:
+    void notify(Notification);
+
 public slots:
     virtual void revisionChanged() Q_DECL_OVERRIDE;
 
@@ -115,23 +119,30 @@ protected:
     struct SyncRequest {
         enum RequestType {
             Synchronization,
-            ChangeReplay
+            ChangeReplay,
+            Flush
         };
 
         SyncRequest(const Sink::QueryBase &q)
-            : flushQueue(false),
-            requestType(Synchronization),
+            : requestType(Synchronization),
             query(q)
         {
         }
 
         SyncRequest(RequestType type)
-            : flushQueue(false),
+            : requestType(type)
+        {
+        }
+
+        SyncRequest(RequestType type, int flushType_, const QByteArray &flushId_)
+            : flushType(flushType_),
+            flushId(flushId_),
             requestType(type)
         {
         }
 
-        bool flushQueue;
+        int flushType = 0;
+        QByteArray flushId;
         RequestType requestType;
         Sink::QueryBase query;
     };
diff --git a/tests/testimplementations.h b/tests/testimplementations.h
index 111c884..6fe08f7 100644
--- a/tests/testimplementations.h
+++ b/tests/testimplementations.h
@@ -66,10 +66,6 @@ public:
     {
         return KAsync::null<void>();
     }
-    KAsync::Job<void> synchronizeResource(bool remoteSync, bool localSync) Q_DECL_OVERRIDE
-    {
-        return KAsync::null<void>();
-    }
     KAsync::Job<void> synchronizeResource(const Sink::QueryBase &) Q_DECL_OVERRIDE
     {
         return KAsync::null<void>();