3 files changed, 32 insertions, 18 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;