13 files changed, 138 insertions, 25 deletions

diff --git a/common/clientapi.cpp b/common/clientapi.cpp
index deab962..40257bb 100644
--- a/common/clientapi.cpp
+++ b/common/clientapi.cpp
@@ -298,13 +298,50 @@ KAsync::Job<void> Resources::inspect(const Inspection &inspectionCommand)
     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]() {
-            //TODO wait for inspection notification
+        .template then<void>([resourceAccess, notifier, id](KAsync::Future<void> &future) {
+            notifier->registerHandler([&future, id](const ResourceNotification &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 ResourceNotification &)> > 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 ResourceNotification &notification) {
+        for (const auto &handler : d->handler) {
+            handler(notification);
+        }
+    });
+    d->resourceAccess << resourceAccess;
+}
+
+void Notifier::registerHandler(std::function<void(const ResourceNotification &)> 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); \
diff --git a/common/clientapi.h b/common/clientapi.h
index d496715..5ed99e0 100644
--- a/common/clientapi.h
+++ b/common/clientapi.h
@@ -31,6 +31,8 @@
 class QAbstractItemModel;
 
 namespace Akonadi2 {
+class ResourceAccess;
+class ResourceNotification;
 
 /**
  * Store interface used in the client API.
@@ -136,6 +138,15 @@ namespace Resources {
     KAsync::Job<void> inspect(const Inspection &inspectionCommand);
 }
 
+class Notifier {
+public:
+    Notifier(const QSharedPointer<ResourceAccess> &resourceAccess);
+    void registerHandler(std::function<void(const ResourceNotification &)>);
+
+private:
+    class Private;
+    QScopedPointer<Private> d;
+};
 
 }
 
diff --git a/common/commands/notification.fbs b/common/commands/notification.fbs
index eb00986..5c810cf 100644
--- a/common/commands/notification.fbs
+++ b/common/commands/notification.fbs
@@ -5,6 +5,7 @@ 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
 }
diff --git a/common/genericresource.cpp b/common/genericresource.cpp
index 90fc763..892c3db 100644
--- a/common/genericresource.cpp
+++ b/common/genericresource.cpp
@@ -301,6 +301,7 @@ GenericResource::GenericResource(const QByteArray &resourceInstanceIdentifier, c
         if (Akonadi2::Commands::VerifyInspectionBuffer(verifier)) {
             auto buffer = Akonadi2::Commands::GetInspection(command);
             int inspectionType = buffer->type();
+            QByteArray inspectionId = QByteArray::fromRawData(reinterpret_cast<const char *>(buffer->id()->Data()), buffer->id()->size());
             QByteArray entityId = QByteArray::fromRawData(reinterpret_cast<const char *>(buffer->entityId()->Data()), buffer->entityId()->size());
             QByteArray domainType = QByteArray::fromRawData(reinterpret_cast<const char *>(buffer->domainType()->Data()), buffer->domainType()->size());
             QByteArray property = QByteArray::fromRawData(reinterpret_cast<const char *>(buffer->property()->Data()), buffer->property()->size());
@@ -308,7 +309,21 @@ GenericResource::GenericResource(const QByteArray &resourceInstanceIdentifier, c
             QDataStream s(expectedValueString);
             QVariant expectedValue;
             s >> expectedValue;
-            return inspect(inspectionType, domainType, entityId, property, expectedValue);
+            inspect(inspectionType, inspectionId, domainType, entityId, property, expectedValue).then<void>([=]() {
+                Akonadi2::ResourceNotification n;
+                n.type = Akonadi2::NotificationType_Inspection;
+                n.id = inspectionId;
+                n.code = Akonadi2::NotificationCode_Success;
+                emit notify(n);
+            }, [=](int code, const QString &message) {
+                Akonadi2::ResourceNotification n;
+                n.type = Akonadi2::NotificationType_Inspection;
+                n.message = message;
+                n.id = inspectionId;
+                n.code = Akonadi2::NotificationCode_Failure;
+                emit notify(n);
+            }).exec();
+            return KAsync::null<void>();
         }
         return KAsync::error<void>(-1, "Invalid inspection command.");
     });
@@ -334,7 +349,7 @@ GenericResource::~GenericResource()
     delete mSourceChangeReplay;
 }
 
-KAsync::Job<void> GenericResource::inspect(int inspectionType, const QByteArray &domainType, const QByteArray &entityId, const QByteArray &property, const QVariant &expectedValue)
+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>();
diff --git a/common/genericresource.h b/common/genericresource.h
index 90b7c29..d71061c 100644
--- a/common/genericresource.h
+++ b/common/genericresource.h
@@ -48,7 +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 &domainType, const QByteArray &entityId, const QByteArray &property, const QVariant &expectedValue);
+    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/listener.cpp b/common/listener.cpp
index 5d55202..bb0a130 100644
--- a/common/listener.cpp
+++ b/common/listener.cpp
@@ -386,6 +386,25 @@ void Listener::updateClientsWithRevision(qint64 revision)
     m_fbb.Clear();
 }
 
+void Listener::notify(const Akonadi2::ResourceNotification &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::NotificationBuilder builder(m_fbb);
+    builder.add_type(static_cast<Akonadi2::NotificationType>(notification.type));
+    builder.add_code(notification.code);
+    builder.add_identifier(idString);
+    builder.add_message(messageString);
+    auto command = builder.Finish();
+    Akonadi2::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) {
@@ -395,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 3db1937..2dfd91a 100644
--- a/common/listener.h
+++ b/common/listener.h
@@ -28,6 +28,7 @@
 namespace Akonadi2
 {
     class Resource;
+    class ResourceNotification;
 }
 
 class QTimer;
@@ -76,6 +77,7 @@ private Q_SLOTS:
     void onDataAvailable();
     void processClientBuffers();
     void refreshRevision(qint64);
+    void notify(const Akonadi2::ResourceNotification &);
     void quit();
 
 private:
diff --git a/common/notification.h b/common/notification.h
index 8ffc24c..c83e579 100644
--- a/common/notification.h
+++ b/common/notification.h
@@ -32,6 +32,7 @@ namespace Akonadi2
 class AKONADI2COMMON_EXPORT ResourceNotification
 {
 public:
+    QByteArray id;
     int type;
     QString message;
     int code;
diff --git a/common/resourceaccess.cpp b/common/resourceaccess.cpp
index 65e9a8c..483d83f 100644
--- a/common/resourceaccess.cpp
+++ b/common/resourceaccess.cpp
@@ -46,6 +46,16 @@
 #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)
+{
+    QTimer *timer = new QTimer;
+    QObject::connect(timer, &QTimer::timeout, [=]() {
+        f();
+        delete timer;
+    });
+    timer->start(0);
+}
+
 namespace Akonadi2
 {
 
@@ -534,6 +544,23 @@ bool ResourceAccess::processMessageBuffer()
                     Log(d->resourceInstanceIdentifier) << "Received shutdown notification.";
                     close();
                     break;
+                case Akonadi2::NotificationType::NotificationType_Inspection: {
+                    Log(d->resourceInstanceIdentifier) << "Received inspection notification.";
+                    ResourceNotification n;
+                    if (buffer->identifier()) {
+                        n.id = QByteArray::fromRawData(reinterpret_cast<char const *>(buffer->identifier()->Data()), buffer->identifier()->size());
+                    }
+                    if (buffer->message()) {
+                        n.message = QByteArray::fromRawData(reinterpret_cast<char const *>(buffer->message()->Data()), buffer->message()->size());
+                    }
+                    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);
+                    });
+                }
+                    break;
                 default:
                     Warning() << "Received unknown notification: " << buffer->type();
                     break;
diff --git a/examples/dummyresource/resourcefactory.cpp b/examples/dummyresource/resourcefactory.cpp
index 27d5f17..c43b5e6 100644
--- a/examples/dummyresource/resourcefactory.cpp
+++ b/examples/dummyresource/resourcefactory.cpp
@@ -134,21 +134,17 @@ void DummyResource::removeFromDisk(const QByteArray &instanceIdentifier)
     Akonadi2::Storage(Akonadi2::storageLocation(), instanceIdentifier + ".synchronization", Akonadi2::Storage::ReadWrite).removeFromDisk();
 }
 
-KAsync::Job<void> DummyResource::inspect(int inspectionType, const QByteArray &domainType, const QByteArray &entityId, const QByteArray &property, const QVariant &expectedValue)
+KAsync::Job<void> DummyResource::inspect(int inspectionType, const QByteArray &inspectionId, const QByteArray &domainType, const QByteArray &entityId, const QByteArray &property, const QVariant &expectedValue)
 {
 
     Trace() << "Inspecting " << inspectionType << domainType << entityId << property << expectedValue;
     if (property == "testInspection") {
-        Akonadi2::ResourceNotification n;
-        n.type = Akonadi2::NotificationType_Inspection;
         if (expectedValue.toBool()) {
             //Success
-            n.code = 0;
-            emit notify(n);
+            return KAsync::null<void>();
         } else {
             //Failure
-            n.code = 1;
-            emit notify(n);
+            return KAsync::error<void>(1, "Failed.");
         }
     }
     return KAsync::null<void>();
diff --git a/examples/dummyresource/resourcefactory.h b/examples/dummyresource/resourcefactory.h
index 3f67187..634829e 100644
--- a/examples/dummyresource/resourcefactory.h
+++ b/examples/dummyresource/resourcefactory.h
@@ -40,7 +40,7 @@ public:
     KAsync::Job<void> synchronizeWithSource(Akonadi2::Storage &mainStore, Akonadi2::Storage &synchronizationStore) Q_DECL_OVERRIDE;
     using GenericResource::synchronizeWithSource;
     static void removeFromDisk(const QByteArray &instanceIdentifier);
-    KAsync::Job<void> inspect(int inspectionType, const QByteArray &domainType, const QByteArray &entityId, const QByteArray &property, const QVariant &expectedValue) Q_DECL_OVERRIDE;
+    KAsync::Job<void> inspect(int inspectionType, const QByteArray &inspectionId, const QByteArray &domainType, const QByteArray &entityId, const QByteArray &property, const QVariant &expectedValue) Q_DECL_OVERRIDE;
 private:
     KAsync::Job<void> replay(Akonadi2::Storage &synchronizationStore, const QByteArray &type, const QByteArray &key, const QByteArray &value) Q_DECL_OVERRIDE;
     Akonadi2::ApplicationDomain::Event::Ptr createEvent(const QByteArray &rid, const QMap<QString, QVariant> &data, Akonadi2::Storage::Transaction &);
diff --git a/tests/clientapitest.cpp b/tests/clientapitest.cpp
index 5942849..86150ee 100644
--- a/tests/clientapitest.cpp
+++ b/tests/clientapitest.cpp
@@ -33,6 +33,7 @@ public:
     KAsync::Job<void> create(const T &domainObject) Q_DECL_OVERRIDE { return KAsync::null<void>(); };
     KAsync::Job<void> modify(const T &domainObject) Q_DECL_OVERRIDE { return KAsync::null<void>(); };
     KAsync::Job<void> remove(const T &domainObject) Q_DECL_OVERRIDE { return KAsync::null<void>(); };
+    KAsync::Job<void> inspect(const Akonadi2::Inspection &) Q_DECL_OVERRIDE { return KAsync::null<void>(); };
     QPair<KAsync::Job<void>, typename Akonadi2::ResultEmitter<typename T::Ptr>::Ptr > load(const Akonadi2::Query &query) Q_DECL_OVERRIDE
     {
         auto resultProvider = new Akonadi2::ResultProvider<typename T::Ptr>();
diff --git a/tests/inspectiontest.cpp b/tests/inspectiontest.cpp
index e332844..29cce6c 100644
--- a/tests/inspectiontest.cpp
+++ b/tests/inspectiontest.cpp
@@ -4,11 +4,7 @@
 
 #include "dummyresource/resourcefactory.h"
 #include "clientapi.h"
-#include "commands.h"
-#include "entitybuffer.h"
 #include "resourceconfig.h"
-#include "modelresult.h"
-#include "pipeline.h"
 #include "log.h"
 
 /**
@@ -38,25 +34,30 @@ private Q_SLOTS:
         Akonadi2::Store::start(QByteArray("org.kde.dummy.instance1")).exec().waitForFinished();
     }
 
-    void init()
+    void testInspection_data()
     {
-        qDebug();
-        qDebug() << "-----------------------------------------";
-        qDebug();
+        QTest::addColumn<bool>("success");
+        QTest::newRow("success") << true;
+        QTest::newRow("fail") << false;
     }
 
-    void testMarkMailAsRead()
+    void testInspection()
     {
+        QFETCH(bool, success);
         using namespace Akonadi2;
         using namespace Akonadi2::ApplicationDomain;
 
         Mail mail(QByteArray("org.kde.dummy.instance1"), QByteArray("identifier"), 0, QSharedPointer<MemoryBufferAdaptor::MemoryBufferAdaptor>::create());
 
-        auto inspectionCommand = Resources::Inspection::PropertyInspection(mail, "unread", true);
+        //testInspection is a magic property that the dummyresource supports
+        auto inspectionCommand = Resources::Inspection::PropertyInspection(mail, "testInspection", success);
         auto result = Resources::inspect<Mail>(inspectionCommand).exec();
         result.waitForFinished();
-        QVERIFY(!result.errorCode());
-        Akonadi2::Store::flushMessageQueue(QByteArrayList() << QByteArray("org.kde.dummy.instance1")).exec().waitForFinished();
+        if (success) {
+            QVERIFY(!result.errorCode());
+        } else {
+            QVERIFY(result.errorCode());
+        }
     }
 };