DomainTypeAdaptor factory, per type preprocessor pipeline configuration.

7 files changed, 77 insertions, 18 deletions

diff --git a/common/entitybuffer.cpp b/common/entitybuffer.cpp
index b9c9d76..c5d6bce 100644
--- a/common/entitybuffer.cpp
+++ b/common/entitybuffer.cpp
@@ -18,6 +18,11 @@ EntityBuffer::EntityBuffer(void *dataValue, int dataSize)
     }
 }
 
+const Akonadi2::Entity &EntityBuffer::entity()
+{
+    return *mEntity;
+}
+
 const flatbuffers::Vector<uint8_t>* EntityBuffer::resourceBuffer()
 {
     if (!mEntity) {
diff --git a/common/entitybuffer.h b/common/entitybuffer.h
index c072777..bd9360d 100644
--- a/common/entitybuffer.h
+++ b/common/entitybuffer.h
@@ -12,6 +12,7 @@ public:
     const flatbuffers::Vector<uint8_t> *resourceBuffer();
     const flatbuffers::Vector<uint8_t> *metadataBuffer();
     const flatbuffers::Vector<uint8_t> *localBuffer();
+    const Entity &entity();
 
     static void extractResourceBuffer(void *dataValue, int dataSize, const std::function<void(const flatbuffers::Vector<uint8_t> *)> &handler);
     static void assembleEntityBuffer(flatbuffers::FlatBufferBuilder &fbb, void *metadataData, size_t metadataSize, void *resourceData, size_t resourceSize, void *localData, size_t localSize);
diff --git a/common/metadata.fbs b/common/metadata.fbs
index 71684b6..34a8df2 100644
--- a/common/metadata.fbs
+++ b/common/metadata.fbs
@@ -2,6 +2,7 @@ namespace Akonadi2;
 
 table Metadata {
     revision: ulong;
+    processed: bool = true;
 }
 
 root_type Metadata;
diff --git a/common/pipeline.cpp b/common/pipeline.cpp
index 04954ac..8d00480 100644
--- a/common/pipeline.cpp
+++ b/common/pipeline.cpp
@@ -41,10 +41,10 @@ public:
     }
 
     Storage storage;
-    QVector<Preprocessor *> nullPipeline;
-    QVector<Preprocessor *> newPipeline;
-    QVector<Preprocessor *> modifiedPipeline;
-    QVector<Preprocessor *> deletedPipeline;
+    QHash<QString, QVector<Preprocessor *> > nullPipeline;
+    QHash<QString, QVector<Preprocessor *> > newPipeline;
+    QHash<QString, QVector<Preprocessor *> > modifiedPipeline;
+    QHash<QString, QVector<Preprocessor *> > deletedPipeline;
     QVector<PipelineState> activePipelines;
     bool stepScheduled;
 };
@@ -60,6 +60,23 @@ Pipeline::~Pipeline()
     delete d;
 }
 
+void Pipeline::setPreprocessors(const QString &entityType, Type pipelineType, const QVector<Preprocessor *> &preprocessors)
+{
+    switch (pipelineType) {
+        case NewPipeline:
+            d->newPipeline[entityType] = preprocessors;
+            break;
+        case ModifiedPipeline:
+            d->modifiedPipeline[entityType] = preprocessors;
+            break;
+        case DeletedPipeline:
+            d->deletedPipeline[entityType] = preprocessors;
+            break;
+        default:
+            break;
+    };
+}
+
 Storage &Pipeline::storage() const
 {
     return d->storage;
@@ -68,12 +85,12 @@ Storage &Pipeline::storage() const
 void Pipeline::null()
 {
     //TODO: is there really any need for the null pipeline? if so, it should be doing something ;)
-    PipelineState state(this, NullPipeline, QByteArray(), d->nullPipeline);
-    d->activePipelines << state;
-    state.step();
+    // PipelineState state(this, NullPipeline, QByteArray(), d->nullPipeline);
+    // d->activePipelines << state;
+    // state.step();
 }
 
-void Pipeline::newEntity(const QByteArray &key, void *resourceBufferData, size_t size)
+void Pipeline::newEntity(const QString &entityType, const QByteArray &key, void *resourceBufferData, size_t size)
 {
     const qint64 newRevision = storage().maxRevision() + 1;
 
@@ -81,6 +98,7 @@ void Pipeline::newEntity(const QByteArray &key, void *resourceBufferData, size_t
     flatbuffers::FlatBufferBuilder metadataFbb;
     auto metadataBuilder = Akonadi2::MetadataBuilder(metadataFbb);
     metadataBuilder.add_revision(newRevision);
+    metadataBuilder.add_processed(false);
     auto metadataBuffer = metadataBuilder.Finish();
     Akonadi2::FinishMetadataBuffer(metadataFbb, metadataBuffer);
 
@@ -90,21 +108,21 @@ void Pipeline::newEntity(const QByteArray &key, void *resourceBufferData, size_t
     storage().write(key.data(), key.size(), fbb.GetBufferPointer(), fbb.GetSize());
     storage().setMaxRevision(newRevision);
 
-    PipelineState state(this, NewPipeline, key, d->newPipeline);
+    PipelineState state(this, NewPipeline, key, d->newPipeline[entityType]);
     d->activePipelines << state;
     state.step();
 }
 
-void Pipeline::modifiedEntity(const QByteArray &key, void *data, size_t size)
+void Pipeline::modifiedEntity(const QString &entityType, const QByteArray &key, void *data, size_t size)
 {
-    PipelineState state(this, ModifiedPipeline, key, d->modifiedPipeline);
+    PipelineState state(this, ModifiedPipeline, key, d->modifiedPipeline[entityType]);
     d->activePipelines << state;
     state.step();
 }
 
-void Pipeline::deletedEntity(const QByteArray &key)
+void Pipeline::deletedEntity(const QString &entityType, const QByteArray &key)
 {
-    PipelineState state(this, DeletedPipeline, key, d->deletedPipeline);
+    PipelineState state(this, DeletedPipeline, key, d->deletedPipeline[entityType]);
     d->activePipelines << state;
     state.step();
 }
diff --git a/common/pipeline.h b/common/pipeline.h
index 6ef8703..8373899 100644
--- a/common/pipeline.h
+++ b/common/pipeline.h
@@ -27,6 +27,7 @@
 
 #include <akonadi2common_export.h>
 #include <storage.h>
+#include <clientapi.h> //For domain types
 
 namespace Akonadi2
 {
@@ -46,12 +47,34 @@ public:
 
     Storage &storage() const;
 
+    // template <typename T>
+    // Storage &storage() const;
+
+    template <typename T>
+    void setPreprocessors(Type type, const QVector<Preprocessor *> &preprocessors)
+    {
+        setPreprocessors(Akonadi2::Domain::getTypeName<T>(), type, preprocessors);
+    }
+
     void null();
-    //FIXME We should probably directly provide a DomainTypeAdapter here. The data has already been written and we only need to read it for processing. And we need to read all buffers.
-    void newEntity(const QByteArray &key, void *resourceBufferData, size_t size);
-    //TODO Send local buffer data as well?
-    void modifiedEntity(const QByteArray &key, void *data, size_t size);
-    void deletedEntity(const QByteArray &key);
+
+    template <typename T>
+    void newEntity(const QByteArray &key, void *resourceBufferData, size_t size)
+    {
+        newEntity(Akonadi2::Domain::getTypeName<T>(), key, resourceBufferData, size);
+    }
+
+    template <typename T>
+    void modifiedEntity(const QByteArray &key, void *data, size_t size)
+    {
+        modifiedEntity(Akonadi2::Domain::getTypeName<T>(), key, data, size);
+    }
+
+    template <typename T>
+    void deletedEntity(const QByteArray &key)
+    {
+        deletedEntity(Akonadi2::Domain::getTypeName<T>(), key);
+    }
 
 Q_SIGNALS:
     void revisionUpdated();
@@ -61,6 +84,10 @@ private Q_SLOTS:
     void stepPipelines();
 
 private:
+    void setPreprocessors(const QString &entityType, Type pipelineType, const QVector<Preprocessor *> &preprocessors);
+    void newEntity(const QString &entityType, const QByteArray &key, void *resourceBufferData, size_t size);
+    void modifiedEntity(const QString &entityType, const QByteArray &key, void *data, size_t size);
+    void deletedEntity(const QString &entityType, const QByteArray &key);
     void pipelineStepped(const PipelineState &state);
     void pipelineCompleted(const PipelineState &state);
     void scheduleStep();
diff --git a/common/resource.cpp b/common/resource.cpp
index bba6609..db08c4f 100644
--- a/common/resource.cpp
+++ b/common/resource.cpp
@@ -39,6 +39,11 @@ Resource::~Resource()
     //delete d;
 }
 
+void Resource::configurePipeline(Pipeline *pipeline)
+{
+
+}
+
 void Resource::processCommand(int commandId, const QByteArray &data, uint size, Pipeline *pipeline)
 {
     Q_UNUSED(commandId)
diff --git a/common/resource.h b/common/resource.h
index fb42c1b..52a28a6 100644
--- a/common/resource.h
+++ b/common/resource.h
@@ -36,6 +36,8 @@ public:
     virtual void processCommand(int commandId, const QByteArray &data, uint size, Pipeline *pipeline);
     virtual Async::Job<void> synchronizeWithSource(Pipeline *pipeline);
 
+    virtual void configurePipeline(Pipeline *pipeline);
+
 private:
     class Private;
     Private * const d;