30 files changed, 667 insertions, 93 deletions

diff --git a/common/CMakeLists.txt b/common/CMakeLists.txt
index 87f4898..85cd621 100644
--- a/common/CMakeLists.txt
+++ b/common/CMakeLists.txt
@@ -75,6 +75,7 @@ generate_flatbuffers(
     commands/synchronize
     commands/notification
     commands/revisionreplayed
+    commands/inspection
     domain/event
     domain/mail
     domain/folder
diff --git a/common/bufferutils.h b/common/bufferutils.h
new file mode 100644
index 0000000..b0fb75a
--- /dev/null
+++ b/common/bufferutils.h
@@ -0,0 +1,26 @@
+#pragma once
+
+#include <flatbuffers/flatbuffers.h>
+#include <QByteArray>
+
+namespace Akonadi2 {
+namespace BufferUtils {
+    template<typename T>
+    static QByteArray extractBuffer(const T *data)
+    {
+        return QByteArray::fromRawData(reinterpret_cast<char const *>(data->Data()), data->size());
+    }
+
+    template<typename T>
+    static QByteArray extractBufferCopy(const T *data)
+    {
+        return QByteArray(reinterpret_cast<char const *>(data->Data()), data->size());
+    }
+
+    static QByteArray extractBuffer(const flatbuffers::FlatBufferBuilder &fbb)
+    {
+        return QByteArray::fromRawData(reinterpret_cast<char const *>(fbb.GetBufferPointer()), fbb.GetSize());
+    }
+}
+}
+
diff --git a/common/clientapi.cpp b/common/clientapi.cpp
index e7ca99d..824ef19 100644
--- a/common/clientapi.cpp
+++ b/common/clientapi.cpp
@@ -25,6 +25,7 @@
 #include <QEventLoop>
 #include <QAbstractItemModel>
 #include <QDir>
+#include <QUuid>
 #include <functional>
 #include <memory>
 
@@ -204,7 +205,7 @@ KAsync::Job<void> Store::synchronize(const Akonadi2::Query &query)
         Trace() << "Synchronizing " << resource;
         auto resourceAccess = QSharedPointer<Akonadi2::ResourceAccess>::create(resource);
         resourceAccess->open();
-        resourceAccess->synchronizeResource(query.syncOnDemand, query.processAll).then<void>([&future, resourceAccess]() {
+        resourceAccess->synchronizeResource(true, false).then<void>([&future, resourceAccess]() {
             future.setFinished();
         }).exec();
     })
@@ -212,10 +213,148 @@ KAsync::Job<void> Store::synchronize(const Akonadi2::Query &query)
     .template then<void>([](){});
 }
 
+KAsync::Job<void> Store::flushMessageQueue(const QByteArrayList &resourceIdentifier)
+{
+    Trace() << "flushMessageQueue" << resourceIdentifier;
+    return KAsync::iterate(resourceIdentifier)
+    .template each<void, QByteArray>([](const QByteArray &resource, KAsync::Future<void> &future) {
+        Trace() << "Flushing message queue " << resource;
+        auto resourceAccess = QSharedPointer<Akonadi2::ResourceAccess>::create(resource);
+        resourceAccess->open();
+        resourceAccess->synchronizeResource(false, true).then<void>([&future, resourceAccess]() {
+            future.setFinished();
+        }).exec();
+    })
+    //FIXME JOBAPI this is only required because we don't care about the return value of each (and each shouldn't even have a return value)
+    .template then<void>([](){});
+}
+
+KAsync::Job<void> Store::flushReplayQueue(const QByteArrayList &resourceIdentifier)
+{
+    return flushMessageQueue(resourceIdentifier);
+}
+
+template <class DomainType>
+KAsync::Job<DomainType> Store::fetchOne(const Akonadi2::Query &query)
+{
+    return KAsync::start<DomainType>([query](KAsync::Future<DomainType> &future) {
+        //FIXME We could do this more elegantly if composed jobs would have the correct type (In that case we'd simply return the value from then continuation, and could avoid the outer job entirely)
+        fetch<DomainType>(query, 1)
+            .template then<void, QList<typename DomainType::Ptr> >([&future](const QList<typename DomainType::Ptr> &list){
+                future.setValue(*list.first());
+                future.setFinished();
+            }, [&future](int errorCode, const QString &errorMessage) {
+                future.setError(errorCode, errorMessage);
+                future.setFinished();
+            }).exec();
+    });
+}
+
+template <class DomainType>
+KAsync::Job<QList<typename DomainType::Ptr> > Store::fetchAll(const Akonadi2::Query &query)
+{
+    return fetch<DomainType>(query);
+}
+
+template <class DomainType>
+KAsync::Job<QList<typename DomainType::Ptr> > Store::fetch(const Akonadi2::Query &query, int minimumAmount)
+{
+    auto model = loadModel<DomainType>(query);
+    auto list = QSharedPointer<QList<typename DomainType::Ptr> >::create();
+    auto context = QSharedPointer<QObject>::create();
+    return KAsync::start<QList<typename DomainType::Ptr> >([model, list, context, minimumAmount](KAsync::Future<QList<typename DomainType::Ptr> > &future) {
+        if (model->rowCount() >= 1) {
+            for (int i = 0; i < model->rowCount(); i++) {
+                list->append(model->index(i, 0, QModelIndex()).data(Akonadi2::Store::DomainObjectRole).template value<typename DomainType::Ptr>());
+            }
+        } else {
+            QObject::connect(model.data(), &QAbstractItemModel::rowsInserted, context.data(), [model, &future, list](const QModelIndex &index, int start, int end) {
+                for (int i = start; i <= end; i++) {
+                    list->append(model->index(i, 0, QModelIndex()).data(Akonadi2::Store::DomainObjectRole).template value<typename DomainType::Ptr>());
+                }
+            });
+            QObject::connect(model.data(), &QAbstractItemModel::dataChanged, context.data(), [model, &future, list, minimumAmount](const QModelIndex &, const QModelIndex &, const QVector<int> &roles) {
+                if (roles.contains(ModelResult<DomainType, typename DomainType::Ptr>::ChildrenFetchedRole)) {
+                    if (list->size() < minimumAmount) {
+                        future.setError(1, "Not enough values.");
+                    } else {
+                        future.setValue(*list);
+                    }
+                    future.setFinished();
+                }
+            });
+        }
+        if (model->data(QModelIndex(), ModelResult<DomainType, typename DomainType::Ptr>::ChildrenFetchedRole).toBool()) {
+            if (list->size() < minimumAmount) {
+                future.setError(1, "Not enough values.");
+            } else {
+                future.setValue(*list);
+            }
+            future.setFinished();
+        }
+    });
+}
+
+template <class DomainType>
+KAsync::Job<void> Resources::inspect(const Inspection &inspectionCommand)
+{
+    auto resource = inspectionCommand.resourceIdentifier;
+
+    Trace() << "Sending inspection " << resource;
+    auto resourceAccess = QSharedPointer<Akonadi2::ResourceAccess>::create(resource);
+    resourceAccess->open();
+    auto notifier = QSharedPointer<Akonadi2::Notifier>::create(resourceAccess);
+    auto id = QUuid::createUuid().toByteArray();
+    return resourceAccess->sendInspectionCommand(id, ApplicationDomain::getTypeName<DomainType>(), inspectionCommand.entityIdentifier, inspectionCommand.property, inspectionCommand.expectedValue)
+        .template then<void>([resourceAccess, notifier, id](KAsync::Future<void> &future) {
+            notifier->registerHandler([&future, id](const Notification &notification) {
+                if (notification.id == id) {
+                    if (notification.code) {
+                        future.setError(-1, "Inspection returned an error: " + notification.message);
+                    } else {
+                        future.setFinished();
+                    }
+                }
+            });
+        });
+}
+
+class Akonadi2::Notifier::Private {
+public:
+    Private()
+        : context(new QObject)
+    {
+
+    }
+    QList<QSharedPointer<ResourceAccess> > resourceAccess;
+    QList<std::function<void(const Notification &)> > handler;
+    QSharedPointer<QObject> context;
+};
+
+Notifier::Notifier(const QSharedPointer<ResourceAccess> &resourceAccess)
+    : d(new Akonadi2::Notifier::Private)
+{
+    QObject::connect(resourceAccess.data(), &ResourceAccess::notification, d->context.data(), [this](const Notification &notification) {
+        for (const auto &handler : d->handler) {
+            handler(notification);
+        }
+    });
+    d->resourceAccess << resourceAccess;
+}
+
+void Notifier::registerHandler(std::function<void(const Notification &)> handler)
+{
+    d->handler << handler;
+}
+
 #define REGISTER_TYPE(T) template KAsync::Job<void> Store::remove<T>(const T &domainObject); \
     template KAsync::Job<void> Store::create<T>(const T &domainObject); \
     template KAsync::Job<void> Store::modify<T>(const T &domainObject); \
     template QSharedPointer<QAbstractItemModel> Store::loadModel<T>(Query query); \
+    template KAsync::Job<void> Resources::inspect<T>(const Inspection &); \
+    template KAsync::Job<T> Store::fetchOne<T>(const Query &); \
+    template KAsync::Job<QList<T::Ptr> > Store::fetchAll<T>(const Query &); \
+    template KAsync::Job<QList<T::Ptr> > Store::fetch<T>(const Query &, int); \
 
 REGISTER_TYPE(ApplicationDomain::Event);
 REGISTER_TYPE(ApplicationDomain::Mail);
diff --git a/common/clientapi.h b/common/clientapi.h
index 4e55432..06376c2 100644
--- a/common/clientapi.h
+++ b/common/clientapi.h
@@ -26,11 +26,14 @@
 #include <Async/Async>
 
 #include "query.h"
+#include "inspection.h"
 #include "applicationdomaintype.h"
 
 class QAbstractItemModel;
 
 namespace Akonadi2 {
+class ResourceAccess;
+class Notification;
 
 /**
  * Store interface used in the client API.
@@ -73,6 +76,11 @@ public:
     static KAsync::Job<void> remove(const DomainType &domainObject);
 
     /**
+     * Synchronize data to local cache.
+     */
+    static KAsync::Job<void> synchronize(const Akonadi2::Query &query);
+
+    /**
      * Shutdown resource.
      */
     static KAsync::Job<void> shutdown(const QByteArray &resourceIdentifier);
@@ -87,16 +95,46 @@ public:
     static KAsync::Job<void> start(const QByteArray &resourceIdentifier);
 
     /**
-     * Synchronize data to local cache.
+     * Flushes any pending messages to disk
      */
-    static KAsync::Job<void> synchronize(const Akonadi2::Query &query);
+    static KAsync::Job<void> flushMessageQueue(const QByteArrayList &resourceIdentifier);
+
+    /**
+     * Flushes any pending messages that haven't been replayed to the source.
+     */
+    static KAsync::Job<void> flushReplayQueue(const QByteArrayList &resourceIdentifier);
 
     /**
      * Removes a resource from disk.
      */
     static void removeFromDisk(const QByteArray &resourceIdentifier);
+
+    template <class DomainType>
+    static KAsync::Job<DomainType> fetchOne(const Akonadi2::Query &query);
+
+    template <class DomainType>
+    static KAsync::Job<QList<typename DomainType::Ptr> > fetchAll(const Akonadi2::Query &query);
+
+    template <class DomainType>
+    static KAsync::Job<QList<typename DomainType::Ptr> > fetch(const Akonadi2::Query &query, int minimumAmount = 0);
 };
 
+namespace Resources {
+    template <class DomainType>
+    KAsync::Job<void> inspect(const Inspection &inspectionCommand);
+}
+
+class Notifier {
+public:
+    Notifier(const QSharedPointer<ResourceAccess> &resourceAccess);
+    // Notifier(const QByteArray &resource);
+    // Notifier(const QByteArrayList &resource);
+    void registerHandler(std::function<void(const Notification &)>);
+
+private:
+    class Private;
+    QScopedPointer<Private> d;
+};
 
 }
 
diff --git a/common/commands.cpp b/common/commands.cpp
index 7a0ae23..35dfb13 100644
--- a/common/commands.cpp
+++ b/common/commands.cpp
@@ -33,7 +33,7 @@ QByteArray name(int commandId)
     switch(commandId) {
         case UnknownCommand:
             return "Unknown";
-        case CommandCompletion:
+        case CommandCompletionCommand:
             return "Completion";
         case HandshakeCommand:
             return "Handshake";
@@ -59,6 +59,8 @@ QByteArray name(int commandId)
             return "Ping";
         case RevisionReplayedCommand:
             return "RevisionReplayed";
+        case InspectionCommand:
+            return "Inspection";
         case CustomCommand:
             return "Custom";
     };
diff --git a/common/commands.h b/common/commands.h
index c68ef90..33d5cd7 100644
--- a/common/commands.h
+++ b/common/commands.h
@@ -34,7 +34,7 @@ namespace Commands
 
 enum CommandIds {
     UnknownCommand = 0,
-    CommandCompletion,
+    CommandCompletionCommand,
     HandshakeCommand,
     RevisionUpdateCommand,
     SynchronizeCommand,
@@ -47,6 +47,7 @@ enum CommandIds {
     NotificationCommand,
     PingCommand,
     RevisionReplayedCommand,
+    InspectionCommand,
     CustomCommand = 0xffff
 };
 
diff --git a/common/commands/commandcompletion.fbs b/common/commands/commandcompletion.fbs
index 5330b4f..de7ec14 100644
--- a/common/commands/commandcompletion.fbs
+++ b/common/commands/commandcompletion.fbs
@@ -1,4 +1,4 @@
-namespace Akonadi2;
+namespace Akonadi2.Commands;
 
 table CommandCompletion {
     id: ulong;
diff --git a/common/commands/fetchentity.fbs b/common/commands/fetchentity.fbs
index ddca275..7a1d74d 100644
--- a/common/commands/fetchentity.fbs
+++ b/common/commands/fetchentity.fbs
@@ -1,4 +1,4 @@
-namespace Akonadi2;
+namespace Akonadi2.Commands;
 
 table FetchEntity {
     revision: ulong;
diff --git a/common/commands/handshake.fbs b/common/commands/handshake.fbs
index 52a883a..e824715 100644
--- a/common/commands/handshake.fbs
+++ b/common/commands/handshake.fbs
@@ -1,4 +1,4 @@
-namespace Akonadi2;
+namespace Akonadi2.Commands;
 
 table Handshake {
     name: string;
diff --git a/common/commands/inspection.fbs b/common/commands/inspection.fbs
new file mode 100644
index 0000000..aaae1ae
--- /dev/null
+++ b/common/commands/inspection.fbs
@@ -0,0 +1,12 @@
+namespace Akonadi2.Commands;
+
+table Inspection {
+    id: string;
+    type: int;
+    entityId: string;
+    domainType: string;
+    property: string;
+    expectedValue: string;
+}
+
+root_type Inspection;
diff --git a/common/commands/notification.fbs b/common/commands/notification.fbs
index 6684472..89687cf 100644
--- a/common/commands/notification.fbs
+++ b/common/commands/notification.fbs
@@ -1,9 +1,13 @@
-namespace Akonadi2;
+namespace Akonadi2.Commands;
 
-enum NotificationType : byte { Shutdown = 1, Status, Warning, Progress }
+enum NotificationType : byte { Shutdown = 1, Status, Warning, Progress, Inspection }
+enum NotificationCode : byte { Success = 0, Failure = 1, UserCode }
 
 table Notification {
     type: NotificationType = Status;
+    identifier: string; //An identifier that links back to the something related to the notification (e.g. an entity id or a command id)
+    message: string;
+    code: int = 0; //Of type NotificationCode
 }
 
 root_type Notification;
diff --git a/common/commands/revisionupdate.fbs b/common/commands/revisionupdate.fbs
index 634bcd0..93fbe34 100644
--- a/common/commands/revisionupdate.fbs
+++ b/common/commands/revisionupdate.fbs
@@ -1,4 +1,4 @@
-namespace Akonadi2;
+namespace Akonadi2.Commands;
 
 table RevisionUpdate {
     revision: ulong;
diff --git a/common/commands/synchronize.fbs b/common/commands/synchronize.fbs
index d2d0364..7c3ae9a 100644
--- a/common/commands/synchronize.fbs
+++ b/common/commands/synchronize.fbs
@@ -1,4 +1,4 @@
-namespace Akonadi2;
+namespace Akonadi2.Commands;
 
 table Synchronize {
     sourceSync: bool; //Synchronize with source
diff --git a/common/domain/applicationdomaintype.h b/common/domain/applicationdomaintype.h
index cff0172..44d8743 100644
--- a/common/domain/applicationdomaintype.h
+++ b/common/domain/applicationdomaintype.h
@@ -82,29 +82,34 @@ inline bool operator==(const ApplicationDomainType& lhs, const ApplicationDomain
             && lhs.resourceInstanceIdentifier() == rhs.resourceInstanceIdentifier();
 }
 
-struct Event : public ApplicationDomainType {
-    typedef QSharedPointer<Event> Ptr;
+struct Entity : public ApplicationDomainType {
+    typedef QSharedPointer<Entity> Ptr;
     using ApplicationDomainType::ApplicationDomainType;
 };
 
-struct Todo : public ApplicationDomainType {
+struct Event : public Entity {
+    typedef QSharedPointer<Event> Ptr;
+    using Entity::Entity;
+};
+
+struct Todo : public Entity {
     typedef QSharedPointer<Todo> Ptr;
-    using ApplicationDomainType::ApplicationDomainType;
+    using Entity::Entity;
 };
 
-struct Calendar : public ApplicationDomainType {
+struct Calendar : public Entity {
     typedef QSharedPointer<Calendar> Ptr;
-    using ApplicationDomainType::ApplicationDomainType;
+    using Entity::Entity;
 };
 
-struct Mail : public ApplicationDomainType {
+struct Mail : public Entity {
     typedef QSharedPointer<Mail> Ptr;
-    using ApplicationDomainType::ApplicationDomainType;
+    using Entity::Entity;
 };
 
-struct Folder : public ApplicationDomainType {
+struct Folder : public Entity {
     typedef QSharedPointer<Folder> Ptr;
-    using ApplicationDomainType::ApplicationDomainType;
+    using Entity::Entity;
 };
 
 /**
@@ -155,6 +160,8 @@ class TypeImplementation;
 
 Q_DECLARE_METATYPE(Akonadi2::ApplicationDomain::ApplicationDomainType)
 Q_DECLARE_METATYPE(Akonadi2::ApplicationDomain::ApplicationDomainType::Ptr)
+Q_DECLARE_METATYPE(Akonadi2::ApplicationDomain::Entity)
+Q_DECLARE_METATYPE(Akonadi2::ApplicationDomain::Entity::Ptr)
 Q_DECLARE_METATYPE(Akonadi2::ApplicationDomain::Event)
 Q_DECLARE_METATYPE(Akonadi2::ApplicationDomain::Event::Ptr)
 Q_DECLARE_METATYPE(Akonadi2::ApplicationDomain::Mail)
diff --git a/common/facade.cpp b/common/facade.cpp
index 22ef84a..91021db 100644
--- a/common/facade.cpp
+++ b/common/facade.cpp
@@ -25,6 +25,7 @@
 #include "definitions.h"
 #include "domainadaptor.h"
 #include "queryrunner.h"
+#include "bufferutils.h"
 
 using namespace Akonadi2;
 
@@ -113,7 +114,7 @@ KAsync::Job<void> GenericFacade<DomainType>::create(const DomainType &domainObje
     }
     flatbuffers::FlatBufferBuilder entityFbb;
     mDomainTypeAdaptorFactory->createBuffer(domainObject, entityFbb);
-    return mResourceAccess->sendCreateCommand(bufferTypeForDomainType(), QByteArray::fromRawData(reinterpret_cast<const char*>(entityFbb.GetBufferPointer()), entityFbb.GetSize()));
+    return mResourceAccess->sendCreateCommand(bufferTypeForDomainType(), BufferUtils::extractBuffer(entityFbb));
 }
 
 template<class DomainType>
@@ -125,7 +126,7 @@ KAsync::Job<void> GenericFacade<DomainType>::modify(const DomainType &domainObje
     }
     flatbuffers::FlatBufferBuilder entityFbb;
     mDomainTypeAdaptorFactory->createBuffer(domainObject, entityFbb);
-    return mResourceAccess->sendModifyCommand(domainObject.identifier(), domainObject.revision(), bufferTypeForDomainType(), QByteArrayList(), QByteArray::fromRawData(reinterpret_cast<const char*>(entityFbb.GetBufferPointer()), entityFbb.GetSize()));
+    return mResourceAccess->sendModifyCommand(domainObject.identifier(), domainObject.revision(), bufferTypeForDomainType(), QByteArrayList(), BufferUtils::extractBuffer(entityFbb));
 }
 
 template<class DomainType>
diff --git a/common/genericresource.cpp b/common/genericresource.cpp
index 29acce4..c7f323a 100644
--- a/common/genericresource.cpp
+++ b/common/genericresource.cpp
@@ -6,13 +6,17 @@
 #include "createentity_generated.h"
 #include "modifyentity_generated.h"
 #include "deleteentity_generated.h"
+#include "inspection_generated.h"
+#include "notification_generated.h"
 #include "domainadaptor.h"
 #include "commands.h"
 #include "index.h"
 #include "log.h"
 #include "definitions.h"
+#include "bufferutils.h"
 
 #include <QUuid>
+#include <QDataStream>
 
 static int sBatchSize = 100;
 
@@ -112,6 +116,7 @@ private:
 class CommandProcessor : public QObject
 {
     Q_OBJECT
+    typedef std::function<KAsync::Job<void>(void const *, size_t)> InspectionFunction;
 public:
     CommandProcessor(Akonadi2::Pipeline *pipeline, QList<MessageQueue*> commandQueues)
         : QObject(),
@@ -135,6 +140,11 @@ public:
         mLowerBoundRevision = revision;
     }
 
+    void setInspectionCommand(const InspectionFunction &f)
+    {
+        mInspect = f;
+    }
+
 
 signals:
     void error(int errorCode, const QString &errorMessage);
@@ -176,6 +186,14 @@ private slots:
                 return mPipeline->modifiedEntity(queuedCommand->command()->Data(), queuedCommand->command()->size());
             case Akonadi2::Commands::CreateEntityCommand:
                 return mPipeline->newEntity(queuedCommand->command()->Data(), queuedCommand->command()->size());
+            case Akonadi2::Commands::InspectionCommand:
+                if (mInspect) {
+                    return mInspect(queuedCommand->command()->Data(), queuedCommand->command()->size()).then<qint64>([]() {
+                        return -1;
+                    });
+                } else {
+                    return KAsync::error<qint64>(-1, "Missing inspection command.");
+                }
             default:
                 return KAsync::error<qint64>(-1, "Unhandled command");
         }
@@ -266,6 +284,7 @@ private:
     bool mProcessingLock;
     //The lowest revision we no longer need
     qint64 mLowerBoundRevision;
+    InspectionFunction mInspect;
 };
 
 
@@ -279,6 +298,38 @@ GenericResource::GenericResource(const QByteArray &resourceInstanceIdentifier, c
     mClientLowerBoundRevision(std::numeric_limits<qint64>::max())
 {
     mProcessor = new CommandProcessor(mPipeline.data(), QList<MessageQueue*>() << &mUserQueue << &mSynchronizerQueue);
+    mProcessor->setInspectionCommand([this](void const *command, size_t size) {
+        flatbuffers::Verifier verifier((const uint8_t *)command, size);
+        if (Akonadi2::Commands::VerifyInspectionBuffer(verifier)) {
+            auto buffer = Akonadi2::Commands::GetInspection(command);
+            int inspectionType = buffer->type();
+
+            QByteArray inspectionId = BufferUtils::extractBuffer(buffer->id());
+            QByteArray entityId = BufferUtils::extractBuffer(buffer->entityId());
+            QByteArray domainType = BufferUtils::extractBuffer(buffer->domainType());
+            QByteArray property = BufferUtils::extractBuffer(buffer->property());
+            QByteArray expectedValueString = BufferUtils::extractBuffer(buffer->expectedValue());
+            QDataStream s(expectedValueString);
+            QVariant expectedValue;
+            s >> expectedValue;
+            inspect(inspectionType, inspectionId, domainType, entityId, property, expectedValue).then<void>([=]() {
+                Akonadi2::Notification n;
+                n.type = Akonadi2::Commands::NotificationType_Inspection;
+                n.id = inspectionId;
+                n.code = Akonadi2::Commands::NotificationCode_Success;
+                emit notify(n);
+            }, [=](int code, const QString &message) {
+                Akonadi2::Notification n;
+                n.type = Akonadi2::Commands::NotificationType_Inspection;
+                n.message = message;
+                n.id = inspectionId;
+                n.code = Akonadi2::Commands::NotificationCode_Failure;
+                emit notify(n);
+            }).exec();
+            return KAsync::null<void>();
+        }
+        return KAsync::error<void>(-1, "Invalid inspection command.");
+    });
     QObject::connect(mProcessor, &CommandProcessor::error, [this](int errorCode, const QString &msg) { onProcessorError(errorCode, msg); });
     QObject::connect(mPipeline.data(), &Pipeline::revisionUpdated, this, &Resource::revisionUpdated);
     mSourceChangeReplay = new ChangeReplay(resourceInstanceIdentifier, [this](const QByteArray &type, const QByteArray &key, const QByteArray &value) {
@@ -301,6 +352,12 @@ GenericResource::~GenericResource()
     delete mSourceChangeReplay;
 }
 
+KAsync::Job<void> GenericResource::inspect(int inspectionType, const QByteArray &inspectionId, const QByteArray &domainType, const QByteArray &entityId, const QByteArray &property, const QVariant &expectedValue)
+{
+    Warning() << "Inspection not implemented";
+    return KAsync::null<void>();
+}
+
 void GenericResource::enableChangeReplay(bool enable)
 {
     if (enable) {
@@ -464,7 +521,7 @@ void GenericResource::createEntity(const QByteArray &akonadiId, const QByteArray
     auto delta = Akonadi2::EntityBuffer::appendAsVector(fbb, entityFbb.GetBufferPointer(), entityFbb.GetSize());
     auto location = Akonadi2::Commands::CreateCreateEntity(fbb, entityId, type, delta, replayToSource);
     Akonadi2::Commands::FinishCreateEntityBuffer(fbb, location);
-    callback(QByteArray::fromRawData(reinterpret_cast<char const *>(fbb.GetBufferPointer()), fbb.GetSize()));
+    callback(BufferUtils::extractBuffer(fbb));
 }
 
 void GenericResource::modifyEntity(const QByteArray &akonadiId, qint64 revision, const QByteArray &bufferType, const Akonadi2::ApplicationDomain::ApplicationDomainType &domainObject, DomainTypeAdaptorFactoryInterface &adaptorFactory, std::function<void(const QByteArray &)> callback)
@@ -481,7 +538,7 @@ void GenericResource::modifyEntity(const QByteArray &akonadiId, qint64 revision,
     //TODO removals
     auto location = Akonadi2::Commands::CreateModifyEntity(fbb, revision, entityId, 0, type, delta, replayToSource);
     Akonadi2::Commands::FinishModifyEntityBuffer(fbb, location);
-    callback(QByteArray::fromRawData(reinterpret_cast<char const *>(fbb.GetBufferPointer()), fbb.GetSize()));
+    callback(BufferUtils::extractBuffer(fbb));
 }
 
 void GenericResource::deleteEntity(const QByteArray &akonadiId, qint64 revision, const QByteArray &bufferType, std::function<void(const QByteArray &)> callback)
@@ -494,7 +551,7 @@ void GenericResource::deleteEntity(const QByteArray &akonadiId, qint64 revision,
     auto type = fbb.CreateString(bufferType.toStdString());
     auto location = Akonadi2::Commands::CreateDeleteEntity(fbb, revision, entityId, type, replayToSource);
     Akonadi2::Commands::FinishDeleteEntityBuffer(fbb, location);
-    callback(QByteArray::fromRawData(reinterpret_cast<char const *>(fbb.GetBufferPointer()), fbb.GetSize()));
+    callback(BufferUtils::extractBuffer(fbb));
 }
 
 void GenericResource::recordRemoteId(const QByteArray &bufferType, const QByteArray &localId, const QByteArray &remoteId, Akonadi2::Storage::Transaction &transaction)
diff --git a/common/genericresource.h b/common/genericresource.h
index f47c6f8..d71061c 100644
--- a/common/genericresource.h
+++ b/common/genericresource.h
@@ -48,6 +48,7 @@ public:
     virtual KAsync::Job<void> synchronizeWithSource(Akonadi2::Storage &mainStore, Akonadi2::Storage &synchronizationStore);
     virtual KAsync::Job<void> processAllMessages() Q_DECL_OVERRIDE;
     virtual void setLowerBoundRevision(qint64 revision) Q_DECL_OVERRIDE;
+    virtual KAsync::Job<void> inspect(int inspectionType, const QByteArray &inspectionId, const QByteArray &domainType, const QByteArray &entityId, const QByteArray &property, const QVariant &expectedValue);
 
     int error() const;
 
diff --git a/common/inspection.h b/common/inspection.h
new file mode 100644
index 0000000..ecf5b3d
--- /dev/null
+++ b/common/inspection.h
@@ -0,0 +1,60 @@
+/*
+ * Copyright (C) 2015 Christian Mollekopf <chrigi_1@fastmail.fm>
+ *
+ * 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 <QByteArray>
+#include <QVariant>
+#include "applicationdomaintype.h"
+
+namespace Akonadi2 {
+    namespace Resources {
+
+struct Inspection {
+    static Inspection PropertyInspection(const Akonadi2::ApplicationDomain::Entity &entity, const QByteArray &property, const QVariant &expectedValue)
+    {
+        Inspection inspection;
+        inspection.resourceIdentifier = entity.resourceInstanceIdentifier();
+        inspection.entityIdentifier = entity.identifier();
+        inspection.property = property;
+        inspection.expectedValue = expectedValue;
+        return inspection;
+    }
+
+    static Inspection ExistenceInspection(const Akonadi2::ApplicationDomain::Entity &entity, bool exists)
+    {
+        Inspection inspection;
+        inspection.resourceIdentifier = entity.resourceInstanceIdentifier();
+        inspection.entityIdentifier = entity.identifier();
+        inspection.expectedValue = exists;
+        return inspection;
+    }
+
+    enum Type {
+        PropertyInspectionType,
+        ExistenceInspectionType
+    };
+    QByteArray resourceIdentifier;
+    QByteArray entityIdentifier;
+    QByteArray property;
+    QVariant expectedValue;
+};
+
+    }
+}
diff --git a/common/listener.cpp b/common/listener.cpp
index 1b78f01..fa08472 100644
--- a/common/listener.cpp
+++ b/common/listener.cpp
@@ -31,11 +31,13 @@
 #include "common/synchronize_generated.h"
 #include "common/notification_generated.h"
 #include "common/revisionreplayed_generated.h"
+#include "common/inspection_generated.h"
 
 #include <QLocalServer>
 #include <QLocalSocket>
 #include <QTimer>
 #include <QTime>
+#include <QDataStream>
 
 Listener::Listener(const QByteArray &resourceInstanceIdentifier, QObject *parent)
     : QObject(parent),
@@ -203,13 +205,14 @@ void Listener::processClientBuffers()
     }
 }
 
-void Listener::processCommand(int commandId, uint messageId, const QByteArray &commandBuffer, Client &client, const std::function<void()> &callback)
+void Listener::processCommand(int commandId, uint messageId, const QByteArray &commandBuffer, Client &client, const std::function<void(bool)> &callback)
 {
+    bool success = true;
     switch (commandId) {
         case Akonadi2::Commands::HandshakeCommand: {
             flatbuffers::Verifier verifier((const uint8_t *)commandBuffer.constData(), commandBuffer.size());
-            if (Akonadi2::VerifyHandshakeBuffer(verifier)) {
-                auto buffer = Akonadi2::GetHandshake(commandBuffer.constData());
+            if (Akonadi2::Commands::VerifyHandshakeBuffer(verifier)) {
+                auto buffer = Akonadi2::Commands::GetHandshake(commandBuffer.constData());
                 client.name = buffer->name()->c_str();
             } else {
                 Warning() << "received invalid command";
@@ -218,8 +221,8 @@ void Listener::processCommand(int commandId, uint messageId, const QByteArray &c
         }
         case Akonadi2::Commands::SynchronizeCommand: {
             flatbuffers::Verifier verifier((const uint8_t *)commandBuffer.constData(), commandBuffer.size());
-            if (Akonadi2::VerifySynchronizeBuffer(verifier)) {
-                auto buffer = Akonadi2::GetSynchronize(commandBuffer.constData());
+            if (Akonadi2::Commands::VerifySynchronizeBuffer(verifier)) {
+                auto buffer = Akonadi2::Commands::GetSynchronize(commandBuffer.constData());
                 Log() << QString("\tSynchronize request (id %1) from %2").arg(messageId).arg(client.name);
                 auto timer = QSharedPointer<QTime>::create();
                 timer->start();
@@ -232,7 +235,7 @@ void Listener::processCommand(int commandId, uint messageId, const QByteArray &c
                 }
                 job.then<void>([callback, timer]() {
                     Trace() << "Sync took " << timer->elapsed();
-                    callback();
+                    callback(true);
                 }).exec();
                 return;
             } else {
@@ -240,6 +243,7 @@ void Listener::processCommand(int commandId, uint messageId, const QByteArray &c
             }
             break;
         }
+        case Akonadi2::Commands::InspectionCommand:
         case Akonadi2::Commands::FetchEntityCommand:
         case Akonadi2::Commands::DeleteEntityCommand:
         case Akonadi2::Commands::ModifyEntityCommand:
@@ -273,11 +277,12 @@ void Listener::processCommand(int commandId, uint messageId, const QByteArray &c
                 Log() << QString("\tReceived custom command from %1: ").arg(client.name) << commandId;
                 loadResource()->processCommand(commandId, commandBuffer);
             } else {
+                success = false;
                 ErrorMsg() << QString("\tReceived invalid command from %1: ").arg(client.name) << commandId;
             }
             break;
     }
-    callback();
+    callback(success);
 }
 
 qint64 Listener::lowerBoundRevision()
@@ -298,8 +303,8 @@ qint64 Listener::lowerBoundRevision()
 void Listener::quit()
 {
     //Broadcast shutdown notifications to open clients, so they don't try to restart the resource
-    auto command = Akonadi2::CreateNotification(m_fbb, Akonadi2::NotificationType::NotificationType_Shutdown);
-    Akonadi2::FinishNotificationBuffer(m_fbb, command);
+    auto command = Akonadi2::Commands::CreateNotification(m_fbb, Akonadi2::Commands::NotificationType::NotificationType_Shutdown);
+    Akonadi2::Commands::FinishNotificationBuffer(m_fbb, command);
     for (Client &client : m_connections) {
         if (client.socket && client.socket->isOpen()) {
             Akonadi2::Commands::write(client.socket, ++m_messageId, Akonadi2::Commands::NotificationCommand, m_fbb);
@@ -333,10 +338,10 @@ bool Listener::processClientBuffer(Client &client)
         auto clientName = client.name;
         const QByteArray commandBuffer = client.commandBuffer.left(size);
         client.commandBuffer.remove(0, size);
-        processCommand(commandId, messageId, commandBuffer, client, [this, messageId, commandId, socket, clientName]() {
+        processCommand(commandId, messageId, commandBuffer, client, [this, messageId, commandId, socket, clientName](bool success) {
             Log() << QString("\tCompleted command messageid %1 of type \"%2\" from %3").arg(messageId).arg(QString(Akonadi2::Commands::name(commandId))).arg(clientName);
             if (socket) {
-                sendCommandCompleted(socket.data(), messageId);
+                sendCommandCompleted(socket.data(), messageId, success);
             } else {
                 Log() << QString("Socket became invalid before we could send a response. client: %1").arg(clientName);
             }
@@ -348,15 +353,15 @@ bool Listener::processClientBuffer(Client &client)
     return false;
 }
 
-void Listener::sendCommandCompleted(QLocalSocket *socket, uint messageId)
+void Listener::sendCommandCompleted(QLocalSocket *socket, uint messageId, bool success)
 {
     if (!socket || !socket->isValid()) {
         return;
     }
 
-    auto command = Akonadi2::CreateCommandCompletion(m_fbb, messageId);
-    Akonadi2::FinishCommandCompletionBuffer(m_fbb, command);
-    Akonadi2::Commands::write(socket, ++m_messageId, Akonadi2::Commands::CommandCompletion, m_fbb);
+    auto command = Akonadi2::Commands::CreateCommandCompletion(m_fbb, messageId, success);
+    Akonadi2::Commands::FinishCommandCompletionBuffer(m_fbb, command);
+    Akonadi2::Commands::write(socket, ++m_messageId, Akonadi2::Commands::CommandCompletionCommand, m_fbb);
     m_fbb.Clear();
 }
 
@@ -367,8 +372,8 @@ void Listener::refreshRevision(qint64 revision)
 
 void Listener::updateClientsWithRevision(qint64 revision)
 {
-    auto command = Akonadi2::CreateRevisionUpdate(m_fbb, revision);
-    Akonadi2::FinishRevisionUpdateBuffer(m_fbb, command);
+    auto command = Akonadi2::Commands::CreateRevisionUpdate(m_fbb, revision);
+    Akonadi2::Commands::FinishRevisionUpdateBuffer(m_fbb, command);
 
     for (const Client &client: m_connections) {
         if (!client.socket || !client.socket->isValid()) {
@@ -381,6 +386,25 @@ void Listener::updateClientsWithRevision(qint64 revision)
     m_fbb.Clear();
 }
 
+void Listener::notify(const Akonadi2::Notification &notification)
+{
+    auto messageString = m_fbb.CreateString(notification.message.toUtf8().constData(), notification.message.toUtf8().size());
+    auto idString = m_fbb.CreateString(notification.id.constData(), notification.id.size());
+    Akonadi2::Commands::NotificationBuilder builder(m_fbb);
+    builder.add_type(static_cast<Akonadi2::Commands::NotificationType>(notification.type));
+    builder.add_code(notification.code);
+    builder.add_identifier(idString);
+    builder.add_message(messageString);
+    auto command = builder.Finish();
+    Akonadi2::Commands::FinishNotificationBuffer(m_fbb, command);
+    for (Client &client : m_connections) {
+        if (client.socket && client.socket->isOpen()) {
+            Akonadi2::Commands::write(client.socket, ++m_messageId, Akonadi2::Commands::NotificationCommand, m_fbb);
+        }
+    }
+    m_fbb.Clear();
+}
+
 Akonadi2::Resource *Listener::loadResource()
 {
     if (!m_resource) {
@@ -390,6 +414,8 @@ Akonadi2::Resource *Listener::loadResource()
             Trace() << QString("\tResource: %1").arg((qlonglong)m_resource);
             connect(m_resource, &Akonadi2::Resource::revisionUpdated,
                 this, &Listener::refreshRevision);
+            connect(m_resource, &Akonadi2::Resource::notify,
+                this, &Listener::notify);
         } else {
             ErrorMsg() << "Failed to load resource " << m_resourceName;
             m_resource = new Akonadi2::Resource;
diff --git a/common/listener.h b/common/listener.h
index 248a190..4112a6a 100644
--- a/common/listener.h
+++ b/common/listener.h
@@ -28,6 +28,7 @@
 namespace Akonadi2
 {
     class Resource;
+    class Notification;
 }
 
 class QTimer;
@@ -76,12 +77,13 @@ private Q_SLOTS:
     void onDataAvailable();
     void processClientBuffers();
     void refreshRevision(qint64);
+    void notify(const Akonadi2::Notification &);
     void quit();
 
 private:
-    void processCommand(int commandId, uint messageId, const QByteArray &commandBuffer, Client &client, const std::function<void()> &callback);
+    void processCommand(int commandId, uint messageId, const QByteArray &commandBuffer, Client &client, const std::function<void(bool)> &callback);
     bool processClientBuffer(Client &client);
-    void sendCommandCompleted(QLocalSocket *socket, uint messageId);
+    void sendCommandCompleted(QLocalSocket *socket, uint messageId, bool success);
     void updateClientsWithRevision(qint64);
     Akonadi2::Resource *loadResource();
     void readFromSocket(QLocalSocket *socket);
diff --git a/common/log.cpp b/common/log.cpp
index c33c700..489e1bd 100644
--- a/common/log.cpp
+++ b/common/log.cpp
@@ -96,7 +96,7 @@ static QString colorCommand(QList<int> colorCodes)
     return string;
 }
 
-QByteArray debugLevelName(DebugLevel debugLevel)
+QByteArray Akonadi2::Log::debugLevelName(DebugLevel debugLevel)
 {
     switch (debugLevel) {
         case DebugLevel::Trace:
@@ -114,15 +114,16 @@ QByteArray debugLevelName(DebugLevel debugLevel)
     return QByteArray();
 }
 
-DebugLevel debugLevelFromName(const QByteArray &name)
+DebugLevel Akonadi2::Log::debugLevelFromName(const QByteArray &name)
 {
-    if (name.toLower() == "trace")
+    const QByteArray lowercaseName = name.toLower();
+    if (lowercaseName == "trace")
         return DebugLevel::Trace;
-    if (name.toLower() == "log")
+    if (lowercaseName == "log")
         return DebugLevel::Log;
-    if (name.toLower() == "warning")
+    if (lowercaseName == "warning")
         return DebugLevel::Warning;
-    if (name.toLower() == "error")
+    if (lowercaseName == "error")
         return DebugLevel::Error;
     return DebugLevel::Log;
 }
@@ -132,6 +133,11 @@ void Akonadi2::Log::setDebugOutputLevel(DebugLevel debugLevel)
     qputenv("AKONADI2DEBUGLEVEL", debugLevelName(debugLevel));
 }
 
+Akonadi2::Log::DebugLevel Akonadi2::Log::debugOutputLevel()
+{
+    return debugLevelFromName(qgetenv("AKONADI2DEBUGLEVEL"));
+}
+
 QDebug Akonadi2::Log::debugStream(DebugLevel debugLevel, int line, const char* file, const char* function, const char* debugArea)
 {
     DebugLevel debugOutputLevel = debugLevelFromName(qgetenv("AKONADI2DEBUGLEVEL"));
diff --git a/common/log.h b/common/log.h
index 9db9e8e..e531348 100644
--- a/common/log.h
+++ b/common/log.h
@@ -12,7 +12,11 @@ enum DebugLevel {
     Error
 };
 
+QByteArray debugLevelName(DebugLevel debugLevel);
+DebugLevel debugLevelFromName(const QByteArray &name);
+
 void setDebugOutputLevel(DebugLevel);
+DebugLevel debugOutputLevel();
 
 QDebug debugStream(DebugLevel debugLevel, int line, const char* file, const char* function, const char* debugArea = 0);
 
diff --git a/common/notification.h b/common/notification.h
new file mode 100644
index 0000000..e1b5bff
--- /dev/null
+++ b/common/notification.h
@@ -0,0 +1,40 @@
+/*
+ * 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 <akonadi2common_export.h>
+#include <QString>
+
+namespace Akonadi2
+{
+
+/**
+ * A notification
+ */
+class AKONADI2COMMON_EXPORT Notification
+{
+public:
+    QByteArray id;
+    int type;
+    QString message;
+    int code;
+};
+
+}
diff --git a/common/pipeline.cpp b/common/pipeline.cpp
index 06d8114..a087def 100644
--- a/common/pipeline.cpp
+++ b/common/pipeline.cpp
@@ -34,6 +34,7 @@
 #include "log.h"
 #include "domain/applicationdomaintype.h"
 #include "definitions.h"
+#include "bufferutils.h"
 
 namespace Akonadi2
 {
@@ -119,7 +120,7 @@ Storage &Pipeline::storage() const
 
 void Pipeline::storeNewRevision(qint64 newRevision, const flatbuffers::FlatBufferBuilder &fbb, const QByteArray &bufferType, const QByteArray &uid)
 {
-    d->transaction.openDatabase(bufferType + ".main").write(Akonadi2::Storage::assembleKey(uid, newRevision), QByteArray::fromRawData(reinterpret_cast<char const *>(fbb.GetBufferPointer()), fbb.GetSize()),
+    d->transaction.openDatabase(bufferType + ".main").write(Akonadi2::Storage::assembleKey(uid, newRevision), BufferUtils::extractBuffer(fbb),
         [](const Akonadi2::Storage::Error &error) {
             Warning() << "Failed to write entity";
         }
@@ -285,7 +286,7 @@ KAsync::Job<qint64> Pipeline::modifiedEntity(void const *command, size_t size)
     //Remove deletions
     if (modifyEntity->deletions()) {
         for (const auto &property : *modifyEntity->deletions()) {
-            newObject->setProperty(QByteArray::fromRawData(property->data(), property->size()), QVariant());
+            newObject->setProperty(BufferUtils::extractBuffer(property), QVariant());
         }
     }
 
diff --git a/common/query.h b/common/query.h
index 0d0f382..1df32da 100644
--- a/common/query.h
+++ b/common/query.h
@@ -22,42 +22,120 @@
 #include <QByteArrayList>
 #include <QHash>
 #include <QSet>
+#include "applicationdomaintype.h"
 
 namespace Akonadi2 {
 
 /**
- * A query that matches a set of objects
- * 
- * The query will have to be updated regularly similary to the domain objects.
- * It probably also makes sense to have a domain specific part of the query,
- * such as what properties we're interested in (necessary information for on-demand
- * loading of data).
- *
- * The query defines:
- * * what resources to search
- * * filters on various properties (parent collection, startDate range, ....)
- * * properties we need (for on-demand querying)
- * 
- * syncOnDemand: Execute a source sync before executing the query
- * processAll: Ensure all local messages are processed before querying to guarantee an up-to date dataset.
+ * A query that matches a set of entities.
  */
 class Query
 {
 public:
-    Query() : syncOnDemand(true), processAll(false), liveQuery(false) {}
-    //Could also be a propertyFilter
+    enum Flag {
+        /** Leave the query running an contiously update the result set. */
+        LiveQuery
+    };
+    Q_DECLARE_FLAGS(Flags, Flag)
+
+    static Query PropertyFilter(const QByteArray &key, const QVariant &value)
+    {
+        Query query;
+        query.propertyFilter.insert(key, value);
+        return query;
+    }
+
+    static Query PropertyFilter(const QByteArray &key, const ApplicationDomain::Entity &entity)
+    {
+        return PropertyFilter(key, QVariant::fromValue(entity.identifier()));
+    }
+
+    static Query ResourceFilter(const QByteArray &identifier)
+    {
+        Query query;
+        query.resources.append(identifier);
+        return query;
+    }
+
+    static Query ResourceFilter(const QByteArrayList &identifier)
+    {
+        Query query;
+        query.resources = identifier;
+        return query;
+    }
+
+    static Query ResourceFilter(const ApplicationDomain::AkonadiResource &entity)
+    {
+        return ResourceFilter(entity.identifier());
+    }
+
+    static Query IdentityFilter(const QByteArray &identifier)
+    {
+        Query query;
+        query.ids << identifier;
+        return query;
+    }
+
+    static Query IdentityFilter(const QByteArrayList &identifier)
+    {
+        Query query;
+        query.ids = identifier;
+        return query;
+    }
+
+    static Query IdentityFilter(const ApplicationDomain::Entity &entity)
+    {
+        return IdentityFilter(entity.identifier());
+    }
+
+    static Query RequestedProperties(const QByteArrayList &properties)
+    {
+        Query query;
+        query.requestedProperties = properties;
+        return query;
+    }
+
+    static Query RequestTree(const QByteArray &parentProperty)
+    {
+        Query query;
+        query.parentProperty = parentProperty;
+        return query;
+    }
+
+    Query(Flags flags = Flags())
+    {}
+
+    Query& operator+=(const Query& rhs)
+    {
+        resources += rhs.resources;
+        ids += rhs.ids;
+        for (auto it = rhs.propertyFilter.constBegin(); it != rhs.propertyFilter.constEnd(); it++) {
+            propertyFilter.insert(it.key(), it.value());
+        }
+        requestedProperties += rhs.requestedProperties;
+        parentProperty = rhs.parentProperty;
+        liveQuery = rhs.liveQuery;
+        syncOnDemand = rhs.syncOnDemand;
+        processAll = rhs.processAll;
+        return *this;
+    }
+
+    friend Query operator+(Query lhs, const Query& rhs)
+    {
+        lhs += rhs;
+        return lhs;
+    }
+
     QByteArrayList resources;
-    //Could also be a propertyFilter
     QByteArrayList ids;
-    //Filters to apply
     QHash<QByteArray, QVariant> propertyFilter;
-    //Properties to retrieve
     QByteArrayList requestedProperties;
     QByteArray parentProperty;
+    bool liveQuery;
     bool syncOnDemand;
     bool processAll;
-    //If live query is false, this query will not continuously be updated
-    bool liveQuery;
 };
 
 }
+
+Q_DECLARE_OPERATORS_FOR_FLAGS(Akonadi2::Query::Flags)
diff --git a/common/resource.h b/common/resource.h
index 4ed21b5..2ae71a0 100644
--- a/common/resource.h
+++ b/common/resource.h
@@ -17,10 +17,12 @@
  * 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 <akonadi2common_export.h>
 
 #include <Async/Async>
+#include "notification.h"
 
 namespace Akonadi2
 {
@@ -55,6 +57,7 @@ public:
 
 Q_SIGNALS:
     void revisionUpdated(qint64);
+    void notify(Notification);
 
 private:
     class Private;
diff --git a/common/resourceaccess.cpp b/common/resourceaccess.cpp
index 7be1259..6592699 100644
--- a/common/resourceaccess.cpp
+++ b/common/resourceaccess.cpp
@@ -30,19 +30,33 @@
 #include "common/modifyentity_generated.h"
 #include "common/deleteentity_generated.h"
 #include "common/revisionreplayed_generated.h"
+#include "common/inspection_generated.h"
 #include "common/entitybuffer.h"
+#include "common/bufferutils.h"
 #include "log.h"
 
 #include <QCoreApplication>
 #include <QDebug>
 #include <QDir>
 #include <QProcess>
+#include <QDataStream>
+#include <QBuffer>
 
 #undef Trace
 #define Trace() Akonadi2::Log::debugStream(Akonadi2::Log::DebugLevel::Trace, __LINE__, __FILE__, Q_FUNC_INFO, "ResourceAccess")
 #undef Log
 #define Log(IDENTIFIER) Akonadi2::Log::debugStream(Akonadi2::Log::DebugLevel::Log, __LINE__, __FILE__, Q_FUNC_INFO, "ResourceAccess("+IDENTIFIER+")")
 
+static void queuedInvoke(const std::function<void()> &f, QObject *context = 0)
+{
+    auto timer = QSharedPointer<QTimer>::create();
+    timer->setSingleShot(true);
+    QObject::connect(timer.data(), &QTimer::timeout, context, [f, timer]() {
+        f();
+    });
+    timer->start(0);
+}
+
 namespace Akonadi2
 {
 
@@ -284,8 +298,8 @@ KAsync::Job<void> ResourceAccess::synchronizeResource(bool sourceSync, bool loca
 {
     Trace() << "Sending synchronize command: " << sourceSync << localSync;
     flatbuffers::FlatBufferBuilder fbb;
-    auto command = Akonadi2::CreateSynchronize(fbb, sourceSync, localSync);
-    Akonadi2::FinishSynchronizeBuffer(fbb, command);
+    auto command = Akonadi2::Commands::CreateSynchronize(fbb, sourceSync, localSync);
+    Akonadi2::Commands::FinishSynchronizeBuffer(fbb, command);
     open();
     return sendCommand(Commands::SynchronizeCommand, fbb);
 }
@@ -338,6 +352,25 @@ KAsync::Job<void> ResourceAccess::sendRevisionReplayedCommand(qint64 revision)
     return sendCommand(Akonadi2::Commands::RevisionReplayedCommand, fbb);
 }
 
+KAsync::Job<void> ResourceAccess::sendInspectionCommand(const QByteArray &inspectionId, const QByteArray &domainType, const QByteArray &entityId, const QByteArray &property, const QVariant &expectedValue)
+{
+    flatbuffers::FlatBufferBuilder fbb;
+    auto id = fbb.CreateString(inspectionId.toStdString());
+    auto domain = fbb.CreateString(domainType.toStdString());
+    auto entity = fbb.CreateString(entityId.toStdString());
+    auto prop = fbb.CreateString(property.toStdString());
+
+    QByteArray array;
+    QDataStream s(&array, QIODevice::WriteOnly);
+    s << expectedValue;
+
+    auto expected = fbb.CreateString(array.toStdString());
+    auto location = Akonadi2::Commands::CreateInspection (fbb, id, 0, entity, domain, prop, expected);
+    Akonadi2::Commands::FinishInspectionBuffer(fbb, location);
+    open();
+    return sendCommand(Akonadi2::Commands::InspectionCommand, fbb);
+}
+
 void ResourceAccess::open()
 {
     if (d->socket && d->socket->isValid()) {
@@ -424,8 +457,8 @@ void ResourceAccess::connected()
     {
         flatbuffers::FlatBufferBuilder fbb;
         auto name = fbb.CreateString(QString("PID: %1 ResourceAccess: %2").arg(QCoreApplication::applicationPid()).arg(reinterpret_cast<qlonglong>(this)).toLatin1());
-        auto command = Akonadi2::CreateHandshake(fbb, name);
-        Akonadi2::FinishHandshakeBuffer(fbb, command);
+        auto command = Akonadi2::Commands::CreateHandshake(fbb, name);
+        Akonadi2::Commands::FinishHandshakeBuffer(fbb, command);
         Commands::write(d->socket.data(), ++d->messageId, Commands::HandshakeCommand, fbb);
     }
 
@@ -490,28 +523,49 @@ bool ResourceAccess::processMessageBuffer()
 
     switch (commandId) {
         case Commands::RevisionUpdateCommand: {
-            auto buffer = GetRevisionUpdate(d->partialMessageBuffer.constData() + headerSize);
+            auto buffer = Commands::GetRevisionUpdate(d->partialMessageBuffer.constData() + headerSize);
             log(QString("Revision updated to: %1").arg(buffer->revision()));
             emit revisionChanged(buffer->revision());
 
             break;
         }
-        case Commands::CommandCompletion: {
-            auto buffer = GetCommandCompletion(d->partialMessageBuffer.constData() + headerSize);
+        case Commands::CommandCompletionCommand: {
+            auto buffer = Commands::GetCommandCompletion(d->partialMessageBuffer.constData() + headerSize);
             log(QString("Command with messageId %1 completed %2").arg(buffer->id()).arg(buffer->success() ? "sucessfully" : "unsuccessfully"));
 
             d->completeCommands << buffer->id();
             //The callbacks can result in this object getting destroyed directly, so we need to ensure we finish our work first
-            QMetaObject::invokeMethod(this, "callCallbacks", Qt::QueuedConnection);
+            queuedInvoke([=]() {
+                d->callCallbacks();
+            }, this);
             break;
         }
         case Commands::NotificationCommand: {
-            auto buffer = GetNotification(d->partialMessageBuffer.constData() + headerSize);
+            auto buffer = Commands::GetNotification(d->partialMessageBuffer.constData() + headerSize);
             switch (buffer->type()) {
-                case Akonadi2::NotificationType::NotificationType_Shutdown:
+                case Akonadi2::Commands::NotificationType::NotificationType_Shutdown:
                     Log(d->resourceInstanceIdentifier) << "Received shutdown notification.";
                     close();
                     break;
+                case Akonadi2::Commands::NotificationType::NotificationType_Inspection: {
+                    Log(d->resourceInstanceIdentifier) << "Received inspection notification.";
+                    Notification n;
+                    if (buffer->identifier()) {
+                        //Don't use fromRawData, the buffer is gone once we invoke emit notification
+                        n.id = BufferUtils::extractBufferCopy(buffer->identifier());
+                    }
+                    if (buffer->message()) {
+                        //Don't use fromRawData, the buffer is gone once we invoke emit notification
+                        n.message = BufferUtils::extractBufferCopy(buffer->message());
+                    }
+                    n.type = buffer->type();
+                    n.code = buffer->code();
+                    //The callbacks can result in this object getting destroyed directly, so we need to ensure we finish our work first
+                    queuedInvoke([=]() {
+                        emit notification(n);
+                    }, this);
+                }
+                    break;
                 default:
                     Warning() << "Received unknown notification: " << buffer->type();
                     break;
@@ -526,11 +580,6 @@ bool ResourceAccess::processMessageBuffer()
     return d->partialMessageBuffer.size() >= headerSize;
 }
 
-void ResourceAccess::callCallbacks()
-{
-    d->callCallbacks();
-}
-
 void ResourceAccess::log(const QString &message)
 {
     Log(d->resourceInstanceIdentifier) << this << message;
diff --git a/common/resourceaccess.h b/common/resourceaccess.h
index 7f61b30..2fe83ed 100644
--- a/common/resourceaccess.h
+++ b/common/resourceaccess.h
@@ -27,6 +27,7 @@
 #include <Async/Async>
 
 #include <flatbuffers/flatbuffers.h>
+#include "notification.h"
 
 namespace Akonadi2
 {
@@ -49,10 +50,12 @@ public:
     virtual KAsync::Job<void> sendModifyCommand(const QByteArray &uid, qint64 revision, const QByteArray &resourceBufferType, const QByteArrayList &deletedProperties, const QByteArray &buffer) { return KAsync::null<void>(); };
     virtual KAsync::Job<void> sendDeleteCommand(const QByteArray &uid, qint64 revision, const QByteArray &resourceBufferType) { return KAsync::null<void>(); };
     virtual KAsync::Job<void> sendRevisionReplayedCommand(qint64 revision) {return KAsync::null<void>(); };
+    virtual KAsync::Job<void> sendInspectionCommand(const QByteArray &inspectionId, const QByteArray &domainType, const QByteArray &entityId, const QByteArray &property, const QVariant &expecedValue) {return KAsync::null<void>(); };
 
 Q_SIGNALS:
     void ready(bool isReady);
     void revisionChanged(qint64 revision);
+    void notification(Notification revision);
 
 public Q_SLOTS:
     virtual void open() = 0;
@@ -78,6 +81,7 @@ public:
     KAsync::Job<void> sendModifyCommand(const QByteArray &uid, qint64 revision, const QByteArray &resourceBufferType, const QByteArrayList &deletedProperties, const QByteArray &buffer) Q_DECL_OVERRIDE;
     KAsync::Job<void> sendDeleteCommand(const QByteArray &uid, qint64 revision, const QByteArray &resourceBufferType) Q_DECL_OVERRIDE;
     KAsync::Job<void> sendRevisionReplayedCommand(qint64 revision) Q_DECL_OVERRIDE;
+    KAsync::Job<void> sendInspectionCommand(const QByteArray &inspectionId, const QByteArray &domainType, const QByteArray &entityId, const QByteArray &property, const QVariant &expecedValue) Q_DECL_OVERRIDE;
     /**
      * Tries to connect to server, and returns a connected socket on success.
      */
@@ -93,7 +97,6 @@ private Q_SLOTS:
     void connectionError(QLocalSocket::LocalSocketError error);
     void readResourceMessage();
     bool processMessageBuffer();
-    void callCallbacks();
 
 private:
     void connected();
diff --git a/common/resourcefacade.h b/common/resourcefacade.h
index 38e0c0e..ae3037a 100644
--- a/common/resourcefacade.h
+++ b/common/resourcefacade.h
@@ -27,6 +27,7 @@
 
 namespace Akonadi2 {
     class Query;
+    class Inspection;
 }
 
 class ResourceFacade : public Akonadi2::StoreFacade<Akonadi2::ApplicationDomain::AkonadiResource>
@@ -38,5 +39,9 @@ public:
     KAsync::Job<void> modify(const Akonadi2::ApplicationDomain::AkonadiResource &resource) Q_DECL_OVERRIDE;
     KAsync::Job<void> remove(const Akonadi2::ApplicationDomain::AkonadiResource &resource) Q_DECL_OVERRIDE;
     QPair<KAsync::Job<void>, typename Akonadi2::ResultEmitter<Akonadi2::ApplicationDomain::AkonadiResource::Ptr>::Ptr > load(const Akonadi2::Query &query) Q_DECL_OVERRIDE;
+    KAsync::Job<void> inspect(const Akonadi2::Inspection &domainObject) Q_DECL_OVERRIDE
+    {
+        return KAsync::error<void>(-1, "Failed to inspect.");
+    }
 };
 
diff --git a/common/resultprovider.h b/common/resultprovider.h
index 7f090e9..a064ab5 100644
--- a/common/resultprovider.h
+++ b/common/resultprovider.h
@@ -346,8 +346,14 @@ public:
         emitter->onRemoved([this](const DomainType &value) {
             this->remove(value);
         });
-        emitter->onInitialResultSetComplete([this](const DomainType &value) {
-            this->initialResultSetComplete(value);
+        auto ptr = emitter.data();
+        emitter->onInitialResultSetComplete([this, ptr](const DomainType &parent) {
+            auto hashValue = qHash(parent);
+            mInitialResultSetInProgress.remove(hashValue, ptr);
+            //Normally a parent is only in a single resource, except the toplevel (invalid) parent
+            if (!mInitialResultSetInProgress.contains(hashValue)) {
+                this->initialResultSetComplete(parent);
+            }
         });
         emitter->onComplete([this]() {
             this->complete();
@@ -365,6 +371,7 @@ public:
             this->initialResultSetComplete(parent);
         } else {
             for (const auto &emitter : mEmitter) {
+                mInitialResultSetInProgress.insert(qHash(parent), emitter.data());
                 emitter->fetch(parent);
             }
         }
@@ -372,6 +379,7 @@ public:
 
 private:
     QList<typename ResultEmitter<DomainType>::Ptr> mEmitter;
+    QMultiMap<qint64, ResultEmitter<DomainType>*> mInitialResultSetInProgress;
 };