A way to ensure all messages have been processed.

As queries become reactive this should become less important. We can then just wait until
all results become available. For tests it is in either case useful though.

13 files changed, 71 insertions, 22 deletions

diff --git a/common/CMakeLists.txt b/common/CMakeLists.txt
index 3d3a2b7..18bcad0 100644
--- a/common/CMakeLists.txt
+++ b/common/CMakeLists.txt
@@ -7,6 +7,7 @@ generate_flatbuffers(
     commands/handshake
     commands/modifyentity
     commands/revisionupdate
+    commands/synchronize
     domain/event
     entity
     metadata
diff --git a/common/clientapi.h b/common/clientapi.h
index 659ae91..55fbed1 100644
--- a/common/clientapi.h
+++ b/common/clientapi.h
@@ -297,7 +297,7 @@ using namespace async;
 class Query
 {
 public:
-    Query() : syncOnDemand(true) {}
+    Query() : syncOnDemand(true), processAll(false) {}
     //Could also be a propertyFilter
     QStringList resources;
     //Could also be a propertyFilter
@@ -307,6 +307,7 @@ public:
     //Properties to retrieve
     QSet<QString> requestedProperties;
     bool syncOnDemand;
+    bool processAll;
 };
 
 
diff --git a/common/commands/synchronize.fbs b/common/commands/synchronize.fbs
new file mode 100644
index 0000000..d2d0364
--- /dev/null
+++ b/common/commands/synchronize.fbs
@@ -0,0 +1,8 @@
+namespace Akonadi2;
+
+table Synchronize {
+    sourceSync: bool; //Synchronize with source
+    localSync: bool; //Ensure all queues are processed so the local state is up-to date.
+}
+
+root_type Synchronize;
diff --git a/common/resource.cpp b/common/resource.cpp
index db08c4f..e158a40 100644
--- a/common/resource.cpp
+++ b/common/resource.cpp
@@ -60,6 +60,11 @@ Async::Job<void> Resource::synchronizeWithSource(Pipeline *pipeline)
     });
 }
 
+Async::Job<void> Resource::processAllMessages()
+{
+    return Async::null<void>();
+}
+
 class ResourceFactory::Private
 {
 public:
diff --git a/common/resource.h b/common/resource.h
index 52a28a6..bcce229 100644
--- a/common/resource.h
+++ b/common/resource.h
@@ -35,6 +35,7 @@ public:
 
     virtual void processCommand(int commandId, const QByteArray &data, uint size, Pipeline *pipeline);
     virtual Async::Job<void> synchronizeWithSource(Pipeline *pipeline);
+    virtual Async::Job<void> processAllMessages();
 
     virtual void configurePipeline(Pipeline *pipeline);
 
diff --git a/common/resourceaccess.cpp b/common/resourceaccess.cpp
index 73a01ca..5d067c5 100644
--- a/common/resourceaccess.cpp
+++ b/common/resourceaccess.cpp
@@ -25,6 +25,7 @@
 #include "common/commandcompletion_generated.h"
 #include "common/handshake_generated.h"
 #include "common/revisionupdate_generated.h"
+#include "common/synchronize_generated.h"
 
 #include <QCoreApplication>
 #include <QDebug>
@@ -186,9 +187,12 @@ Async::Job<void>  ResourceAccess::sendCommand(int commandId, flatbuffers::FlatBu
     });
 }
 
-Async::Job<void> ResourceAccess::synchronizeResource()
+Async::Job<void> ResourceAccess::synchronizeResource(bool sourceSync, bool localSync)
 {
-    return sendCommand(Commands::SynchronizeCommand);
+    auto command = Akonadi2::CreateSynchronize(d->fbb, sourceSync, localSync);
+    Akonadi2::FinishSynchronizeBuffer(d->fbb, command);
+    return sendCommand(Commands::SynchronizeCommand, d->fbb);
+    d->fbb.Clear();
 }
 
 void ResourceAccess::open()
diff --git a/common/resourceaccess.h b/common/resourceaccess.h
index a9e8c47..fbdd992 100644
--- a/common/resourceaccess.h
+++ b/common/resourceaccess.h
@@ -43,7 +43,7 @@ public:
 
     Async::Job<void> sendCommand(int commandId);
     Async::Job<void> sendCommand(int commandId, flatbuffers::FlatBufferBuilder &fbb);
-    Async::Job<void> synchronizeResource();
+    Async::Job<void> synchronizeResource(bool remoteSync, bool localSync);
 
 public Q_SLOTS:
     void open();
diff --git a/dummyresource/facade.cpp b/dummyresource/facade.cpp
index 1aaec68..fcce871 100644
--- a/dummyresource/facade.cpp
+++ b/dummyresource/facade.cpp
@@ -115,16 +115,16 @@ static std::function<bool(const std::string &key, DummyEvent const *buffer, Akon
     return preparedQuery;
 }
 
-Async::Job<void> DummyResourceFacade::synchronizeResource(bool sync)
+Async::Job<void> DummyResourceFacade::synchronizeResource(bool sync, bool processAll)
 {
     //TODO check if a sync is necessary
     //TODO Only sync what was requested
     //TODO timeout
 
-    if (sync) {
+    if (sync || processAll) {
         return Async::start<void>([=](Async::Future<void> &future) {
             mResourceAccess->open();
-            mResourceAccess->synchronizeResource().then<void>([&future](Async::Future<void> &f) {
+            mResourceAccess->synchronizeResource(sync, processAll).then<void>([&future](Async::Future<void> &f) {
                 future.setFinished();
                 f.setFinished();
             }).exec();
@@ -195,7 +195,7 @@ void DummyResourceFacade::readValue(QSharedPointer<Akonadi2::Storage> storage, c
 
 Async::Job<void> DummyResourceFacade::load(const Akonadi2::Query &query, const std::function<void(const Akonadi2::Domain::Event::Ptr &)> &resultCallback)
 {
-    return synchronizeResource(query.syncOnDemand).then<void>([=](Async::Future<void> &future) {
+    return synchronizeResource(query.syncOnDemand, query.processAll).then<void>([=](Async::Future<void> &future) {
         //Now that the sync is complete we can execute the query
         const auto preparedQuery = prepareQuery(query);
 
diff --git a/dummyresource/facade.h b/dummyresource/facade.h
index c9c8047..1f69161 100644
--- a/dummyresource/facade.h
+++ b/dummyresource/facade.h
@@ -44,7 +44,7 @@ public:
 
 private:
     void readValue(QSharedPointer<Akonadi2::Storage> storage, const QByteArray &key, const std::function<void(const Akonadi2::Domain::Event::Ptr &)> &resultCallback, std::function<bool(const std::string &key, DummyCalendar::DummyEvent const *buffer, Akonadi2::Domain::Buffer::Event const *local)>);
-    Async::Job<void> synchronizeResource(bool sync);
+    Async::Job<void> synchronizeResource(bool sync, bool processAll);
     QSharedPointer<Akonadi2::ResourceAccess> mResourceAccess;
     QSharedPointer<DomainTypeAdaptorFactory<Akonadi2::Domain::Event, Akonadi2::Domain::Buffer::Event, DummyCalendar::DummyEvent> > mFactory;
 };
diff --git a/dummyresource/resourcefactory.cpp b/dummyresource/resourcefactory.cpp
index 18083cb..f510cd5 100644
--- a/dummyresource/resourcefactory.cpp
+++ b/dummyresource/resourcefactory.cpp
@@ -387,7 +387,14 @@ Async::Job<void> DummyResource::synchronizeWithSource(Akonadi2::Pipeline *pipeli
             }
         }
         //TODO find items to remove
+        qDebug() << "sync complete";
+        f.setFinished();
+    });
+}
 
+Async::Job<void> DummyResource::processAllMessages()
+{
+    return Async::start<void>([this](Async::Future<void> &f) {
         //We have to wait for all items to be processed to ensure the synced items are available when a query gets executed.
         //TODO: report errors while processing sync?
         if (mSynchronizerQueue.isEmpty()) {
diff --git a/dummyresource/resourcefactory.h b/dummyresource/resourcefactory.h
index 767eba1..3b99d5e 100644
--- a/dummyresource/resourcefactory.h
+++ b/dummyresource/resourcefactory.h
@@ -35,6 +35,7 @@ class DummyResource : public Akonadi2::Resource
 public:
     DummyResource();
     Async::Job<void> synchronizeWithSource(Akonadi2::Pipeline *pipeline);
+    Async::Job<void> processAllMessages();
     void processCommand(int commandId, const QByteArray &data, uint size, Akonadi2::Pipeline *pipeline);
     void configurePipeline(Akonadi2::Pipeline *pipeline);
     int error() const;
diff --git a/synchronizer/listener.cpp b/synchronizer/listener.cpp
index a84623d..6098856 100644
--- a/synchronizer/listener.cpp
+++ b/synchronizer/listener.cpp
@@ -28,6 +28,7 @@
 #include "common/commandcompletion_generated.h"
 #include "common/handshake_generated.h"
 #include "common/revisionupdate_generated.h"
+#include "common/synchronize_generated.h"
 
 #include <QLocalSocket>
 #include <QTimer>
@@ -189,18 +190,38 @@ void Listener::processCommand(int commandId, uint messageId, Client &client, uin
             break;
         }
         case Akonadi2::Commands::SynchronizeCommand: {
-            log(QString("\tSynchronize request (id %1) from %2").arg(messageId).arg(client.name));
-            loadResource();
-            if (m_resource) {
-                qDebug() << "synchronizing";
-                m_resource->synchronizeWithSource(m_pipeline).then<void>([callback](Async::Future<void> &f){
-                    //FIXME: the command is complete once all pipelines have been drained, track progress and async emit result
-                    callback();
-                    f.setFinished();
-                }).exec();
+            flatbuffers::Verifier verifier((const uint8_t *)client.commandBuffer.constData(), size);
+            if (Akonadi2::VerifySynchronizeBuffer(verifier)) {
+                auto buffer = Akonadi2::GetSynchronize(client.commandBuffer.constData());
+                log(QString("\tSynchronize request (id %1) from %2").arg(messageId).arg(client.name));
+                loadResource();
+                if (!m_resource) {
+                    qWarning() << "No resource loaded";
+                    break;
+                }
+                //TODO a more elegant composition of jobs should be possible
+                if (buffer->sourceSync()) {
+                    bool localSync = buffer->localSync();
+                    m_resource->synchronizeWithSource(m_pipeline).then<void>([callback, localSync, this](Async::Future<void> &f){
+                        if (localSync) {
+                            m_resource->processAllMessages().then<void>([callback](Async::Future<void> &f){
+                                callback();
+                                f.setFinished();
+                            }).exec();
+                        } else {
+                            callback();
+                            f.setFinished();
+                        }
+                    }).exec();
+                } else if (buffer->localSync()) {
+                    m_resource->processAllMessages().then<void>([callback](Async::Future<void> &f){
+                        callback();
+                        f.setFinished();
+                    }).exec();
+                }
                 return;
             } else {
-                qWarning() << "No resource loaded";
+                qWarning() << "received invalid command";
             }
             break;
         }
diff --git a/tests/dummyresourcetest.cpp b/tests/dummyresourcetest.cpp
index 4e9d0a5..5508452 100644
--- a/tests/dummyresourcetest.cpp
+++ b/tests/dummyresourcetest.cpp
@@ -108,12 +108,10 @@ private Q_SLOTS:
         event.setProperty("summary", "summaryValue");
         Akonadi2::Store::create<Akonadi2::Domain::Event>(event, "org.kde.dummy");
 
-        //TODO required to ensure all messages have been processed. The query should ensure this.
-        QTest::qWait(300);
-
         Akonadi2::Query query;
         query.resources << "org.kde.dummy";
         query.syncOnDemand = false;
+        query.processAll = false;
 
         query.propertyFilter.insert("uid", "testuid");
         async::SyncListResult<Akonadi2::Domain::Event::Ptr> result(Akonadi2::Store::load<Akonadi2::Domain::Event>(query));
@@ -141,6 +139,8 @@ private Q_SLOTS:
     {
         Akonadi2::Query query;
         query.resources << "org.kde.dummy";
+        query.syncOnDemand = true;
+        query.processAll = true;
 
         async::SyncListResult<Akonadi2::Domain::Event::Ptr> result(Akonadi2::Store::load<Akonadi2::Domain::Event>(query));
         result.exec();