1 files changed, 60 insertions, 3 deletions

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)