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();