7 files changed, 84 insertions, 50 deletions

diff --git a/common/clientapi.h b/common/clientapi.h
index 2f1c127..dd11a0d 100644
--- a/common/clientapi.h
+++ b/common/clientapi.h
@@ -165,7 +165,7 @@ namespace Akonadi2 {
 /**
  * Standardized Domain Types
  *
- * The don't adhere to any standard and can be freely extended
+ * They don't adhere to any standard and can be freely extended
  * Their sole purpose is providing a standardized interface to access data.
  * 
  * This is necessary to decouple resource-backends from application domain containers (otherwise each resource would have to provide a faceade for each application domain container).
@@ -297,6 +297,7 @@ using namespace async;
 class Query
 {
 public:
+    Query() : syncOnDemand(true) {}
     //Could also be a propertyFilter
     QStringList resources;
     //Could also be a propertyFilter
@@ -305,6 +306,7 @@ public:
     QHash<QString, QVariant> propertyFilter;
     //Properties to retrieve
     QSet<QString> requestedProperties;
+    bool syncOnDemand;
 };
 
 
@@ -324,7 +326,7 @@ public:
     virtual Async::Job<void> create(const DomainType &domainObject) = 0;
     virtual Async::Job<void> modify(const DomainType &domainObject) = 0;
     virtual Async::Job<void> remove(const DomainType &domainObject) = 0;
-    virtual void load(const Query &query, const std::function<void(const typename DomainType::Ptr &)> &resultCallback, const std::function<void()> &completeCallback) = 0;
+    virtual Async::Job<void> load(const Query &query, const std::function<void(const typename DomainType::Ptr &)> &resultCallback) = 0;
 };
 
 
@@ -418,21 +420,25 @@ public:
             // query tells us in which resources we're interested
             // TODO: queries to individual resources could be parallelized
             auto eventloop = QSharedPointer<QEventLoop>::create();
-            int completeCounter = 0;
+            Async::Job<void> job = Async::null<void>();
             for(const QString &resource : query.resources) {
                 auto facade = FacadeFactory::instance().getFacade<DomainType>(resource);
                 //We have to bind an instance to the function callback. Since we use a shared pointer this keeps the result provider instance (and thus also the emitter) alive.
                 std::function<void(const typename DomainType::Ptr &)> addCallback = std::bind(&ResultProvider<typename DomainType::Ptr>::add, resultSet, std::placeholders::_1);
                 //We copy the facade pointer to keep it alive
-                facade->load(query, addCallback, [&completeCounter, &query, resultSet, facade, eventloop]() {
-                    //TODO use jobs instead of this counter
-                    completeCounter++;
-                    if (completeCounter == query.resources.size()) {
-                        resultSet->complete();
-                        eventloop->quit();
-                    }
+                job = job.then<void>([facade, query, addCallback](Async::Future<void> &future) {
+                    Async::Job<void> j = facade->load(query, addCallback);
+                    j.then<void>([&future, facade](Async::Future<void> &f) {
+                        future.setFinished();
+                        f.setFinished();
+                    }).exec();
                 });
             }
+            job.then<void>([eventloop, resultSet](Async::Future<void> &future) {
+                resultSet->complete();
+                eventloop->quit();
+                future.setFinished();
+            }).exec();
             //The thread contains no eventloop, so execute one here
             eventloop->exec(QEventLoop::ExcludeUserInputEvents);
         });
diff --git a/common/pipeline.cpp b/common/pipeline.cpp
index 339a39c..dda7671 100644
--- a/common/pipeline.cpp
+++ b/common/pipeline.cpp
@@ -106,15 +106,21 @@ Async::Job<void> Pipeline::newEntity(void const *command, size_t size)
     {
         flatbuffers::Verifier verifyer(reinterpret_cast<const uint8_t *>(command), size);
         if (!Akonadi2::Commands::VerifyCreateEntityBuffer(verifyer)) {
-            qWarning() << "invalid buffer";
+            qWarning() << "invalid buffer, not a create entity buffer";
             return Async::null<void>();
         }
     }
-
     auto createEntity = Akonadi2::Commands::GetCreateEntity(command);
 
     //TODO rename createEntitiy->domainType to bufferType
     const QString entityType = QString::fromUtf8(reinterpret_cast<char const*>(createEntity->domainType()->Data()), createEntity->domainType()->size());
+    {
+        flatbuffers::Verifier verifyer(reinterpret_cast<const uint8_t *>(createEntity->delta()->Data()), createEntity->delta()->size());
+        if (!Akonadi2::VerifyEntityBuffer(verifyer)) {
+            qWarning() << "invalid buffer, not an entity buffer";
+            return Async::null<void>();
+        }
+    }
     auto entity = Akonadi2::GetEntity(createEntity->delta()->Data());
 
     //Add metadata buffer
diff --git a/common/test/clientapitest.cpp b/common/test/clientapitest.cpp
index 16616a3..c9e4d6d 100644
--- a/common/test/clientapitest.cpp
+++ b/common/test/clientapitest.cpp
@@ -11,13 +11,15 @@ public:
     virtual Async::Job<void> create(const Akonadi2::Domain::Event &domainObject){ return Async::null<void>(); };
     virtual Async::Job<void> modify(const Akonadi2::Domain::Event &domainObject){ return Async::null<void>(); };
     virtual Async::Job<void> remove(const Akonadi2::Domain::Event &domainObject){ return Async::null<void>(); };
-    virtual void load(const Akonadi2::Query &query, const std::function<void(const Akonadi2::Domain::Event::Ptr &)> &resultCallback, const std::function<void()> &completeCallback)
+    virtual Async::Job<void> load(const Akonadi2::Query &query, const std::function<void(const Akonadi2::Domain::Event::Ptr &)> &resultCallback)
     {
-        qDebug() << "load called";
-        for(const auto &result : results) {
-            resultCallback(result);
-        }
-        completeCallback();
+        return Async::start<void>([this, resultCallback](Async::Future<void> &future) {
+            qDebug() << "load called";
+            for(const auto &result : results) {
+                resultCallback(result);
+            }
+            future.setFinished();
+        });
     }
 
     QList<Akonadi2::Domain::Event::Ptr> results;
diff --git a/dummyresource/facade.cpp b/dummyresource/facade.cpp
index 13c174b..b7ba2c2 100644
--- a/dummyresource/facade.cpp
+++ b/dummyresource/facade.cpp
@@ -118,22 +118,29 @@ static std::function<bool(const std::string &key, DummyEvent const *buffer)> pre
     return preparedQuery;
 }
 
-void DummyResourceFacade::synchronizeResource(const std::function<void()> &continuation)
+Async::Job<void> DummyResourceFacade::synchronizeResource(bool sync)
 {
     //TODO check if a sync is necessary
     //TODO Only sync what was requested
     //TODO timeout
-    mResourceAccess->synchronizeResource().then<void>([continuation](Async::Future<void> &f){
-        continuation();
-        f.setFinished();
-    }).exec();
+
+    if (sync) {
+        return Async::start<void>([=](Async::Future<void> &future) {
+            mResourceAccess->open();
+            mResourceAccess->synchronizeResource().then<void>([&future](Async::Future<void> &f) {
+                future.setFinished();
+                f.setFinished();
+            }).exec();
+        });
+    }
+    return Async::null<void>();
 }
 
-void DummyResourceFacade::load(const Akonadi2::Query &query, const std::function<void(const Akonadi2::Domain::Event::Ptr &)> &resultCallback, const std::function<void()> &completeCallback)
+Async::Job<void> DummyResourceFacade::load(const Akonadi2::Query &query, const std::function<void(const Akonadi2::Domain::Event::Ptr &)> &resultCallback)
 {
     qDebug() << "load called";
 
-    synchronizeResource([=]() {
+    return synchronizeResource(query.syncOnDemand).then<void>([=](Async::Future<void> &future) {
         qDebug() << "sync complete";
         //Now that the sync is complete we can execute the query
         const auto preparedQuery = prepareQuery(query);
@@ -188,7 +195,7 @@ void DummyResourceFacade::load(const Akonadi2::Query &query, const std::function
             }
             return true;
         });
-        completeCallback();
+        future.setFinished();
     });
 }
 
diff --git a/dummyresource/facade.h b/dummyresource/facade.h
index 9c8827a..da0b1d6 100644
--- a/dummyresource/facade.h
+++ b/dummyresource/facade.h
@@ -40,10 +40,10 @@ public:
     virtual Async::Job<void> create(const Akonadi2::Domain::Event &domainObject);
     virtual Async::Job<void> modify(const Akonadi2::Domain::Event &domainObject);
     virtual Async::Job<void> remove(const Akonadi2::Domain::Event &domainObject);
-    virtual void load(const Akonadi2::Query &query, const std::function<void(const Akonadi2::Domain::Event::Ptr &)> &resultCallback, const std::function<void()> &completeCallback);
+    virtual Async::Job<void> load(const Akonadi2::Query &query, const std::function<void(const Akonadi2::Domain::Event::Ptr &)> &resultCallback);
 
 private:
-    void synchronizeResource(const std::function<void()> &continuation);
+    Async::Job<void> synchronizeResource(bool sync);
     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 f74eb72..dc716ef 100644
--- a/dummyresource/resourcefactory.cpp
+++ b/dummyresource/resourcefactory.cpp
@@ -317,7 +317,9 @@ Async::Job<void> DummyResource::synchronizeWithSource(Akonadi2::Pipeline *pipeli
                 builder.add_attachment(attachment);
                 auto buffer = builder.Finish();
                 DummyCalendar::FinishDummyEventBuffer(m_fbb, buffer);
-                enqueueCommand(mSynchronizerQueue, Akonadi2::Commands::CreateEntityCommand, QByteArray::fromRawData(reinterpret_cast<char const *>(m_fbb.GetBufferPointer()), m_fbb.GetSize()));
+                flatbuffers::FlatBufferBuilder entityFbb;
+                Akonadi2::EntityBuffer::assembleEntityBuffer(entityFbb, 0, 0, m_fbb.GetBufferPointer(), m_fbb.GetSize(), 0, 0);
+                enqueueCommand(mSynchronizerQueue, Akonadi2::Commands::CreateEntityCommand, QByteArray::fromRawData(reinterpret_cast<char const *>(entityFbb.GetBufferPointer()), entityFbb.GetSize()));
             } else { //modification
                 //TODO diff and create modification if necessary
             }
diff --git a/tests/dummyresourcetest.cpp b/tests/dummyresourcetest.cpp
index c469796..a80d22f 100644
--- a/tests/dummyresourcetest.cpp
+++ b/tests/dummyresourcetest.cpp
@@ -82,27 +82,38 @@ private Q_SLOTS:
         QCOMPARE(revisionSpy.count(), 2);
     }
 
-    // void testResourceSync()
-    // {
-    //     Akonadi2::Pipeline pipeline("org.kde.dummy");
-    //     DummyResource resource;
-    //     auto job = resource.synchronizeWithSource(&pipeline);
-    //     auto future = job.exec();
-    //     QTRY_VERIFY(future.isFinished());
-    // }
-
-    // void testSyncAndFacade()
-    // {
-    //     Akonadi2::Query query;
-    //     query.resources << "org.kde.dummy";
-
-    //     async::SyncListResult<Akonadi2::Domain::Event::Ptr> result(Akonadi2::Store::load<Akonadi2::Domain::Event>(query));
-    //     result.exec();
-    //     QVERIFY(!result.isEmpty());
-    //     auto value = result.first();
-    //     QVERIFY(!value->getProperty("summary").toString().isEmpty());
-    //     qDebug() << value->getProperty("summary").toString();
-    // }
+    void testWriteToFacade()
+    {
+        Akonadi2::Query query;
+        Akonadi2::Domain::Event event;
+        event.setProperty("summary", "summaryValue");
+        Akonadi2::Store::create<Akonadi2::Domain::Event>(event, "org.kde.dummy");
+
+        QTest::qWait(1000);
+        //TODO wait for success response
+    }
+
+    void testResourceSync()
+    {
+        Akonadi2::Pipeline pipeline("org.kde.dummy");
+        DummyResource resource;
+        auto job = resource.synchronizeWithSource(&pipeline);
+        auto future = job.exec();
+        QTRY_VERIFY(future.isFinished());
+    }
+
+    void testSyncAndFacade()
+    {
+        Akonadi2::Query query;
+        query.resources << "org.kde.dummy";
+
+        async::SyncListResult<Akonadi2::Domain::Event::Ptr> result(Akonadi2::Store::load<Akonadi2::Domain::Event>(query));
+        result.exec();
+        QVERIFY(!result.isEmpty());
+        auto value = result.first();
+        QVERIFY(!value->getProperty("summary").toString().isEmpty());
+        qDebug() << value->getProperty("summary").toString();
+    }
 
 };