Made pipeline preprocessing synchronous.

Instead of having the asynchronous preprocessor concept with different
pipelines for new/modify/delete we have a single pipeline with
synchronous preprocessors that act upon new/modify/delete.
This keeps the code simpler due to lack of asynchronity and keeps the
new/modify/delete operations together (which at least for the indexing
makes a lot of sense).
Not supporting asynchronity is ok because the tasks done in
preprocessing are not cpu intensive (if they were we had a problem
since they are directly involved in the round-trip time), and the main
cost comes from i/o, meaning we don't gain much by doing multithreading.
Costly tasks (such as full-text indexing) should rather be implemented
as post-processing, since that doesn't increase the round-trip time directly,
and eventually consistent is typically good enough for that.

12 files changed, 251 insertions, 356 deletions

diff --git a/common/domain/event.cpp b/common/domain/event.cpp
index 9759fc3..83a6906 100644
--- a/common/domain/event.cpp
+++ b/common/domain/event.cpp
@@ -50,11 +50,11 @@ ResultSet TypeImplementation<Event>::queryIndexes(const Akonadi2::Query &query,
     return ResultSet(keys);
 }
 
-void TypeImplementation<Event>::index(const Event &type, Akonadi2::Storage::Transaction &transaction)
+void TypeImplementation<Event>::index(const QByteArray &identifier, const BufferAdaptor &bufferAdaptor, Akonadi2::Storage::Transaction &transaction)
 {
-    const auto uid = type.getProperty("uid");
+    const auto uid = bufferAdaptor.getProperty("uid");
     if (uid.isValid()) {
-        Index("event.index.uid", transaction).add(uid.toByteArray(), type.identifier());
+        Index("event.index.uid", transaction).add(uid.toByteArray(), identifier);
     }
 }
 
diff --git a/common/domain/event.h b/common/domain/event.h
index f21cd34..e9ba52a 100644
--- a/common/domain/event.h
+++ b/common/domain/event.h
@@ -56,7 +56,7 @@ public:
      * An empty result set indicates that a full scan is required.
      */
     static ResultSet queryIndexes(const Akonadi2::Query &query, const QByteArray &resourceInstanceIdentifier, QSet<QByteArray> &appliedFilters, Akonadi2::Storage::Transaction &transaction);
-    static void index(const Event &type, Akonadi2::Storage::Transaction &transaction);
+    static void index(const QByteArray &identifier, const BufferAdaptor &bufferAdaptor, Akonadi2::Storage::Transaction &transaction);
     static QSharedPointer<ReadPropertyMapper<Buffer> > initializeReadPropertyMapper();
     static QSharedPointer<WritePropertyMapper<BufferBuilder> > initializeWritePropertyMapper();
 };
diff --git a/common/domain/mail.cpp b/common/domain/mail.cpp
index d40dde9..ffe322e 100644
--- a/common/domain/mail.cpp
+++ b/common/domain/mail.cpp
@@ -50,12 +50,11 @@ ResultSet TypeImplementation<Mail>::queryIndexes(const Akonadi2::Query &query, c
     return ResultSet(keys);
 }
 
-void TypeImplementation<Mail>::index(const Mail &type, Akonadi2::Storage::Transaction &transaction)
+void TypeImplementation<Mail>::index(const QByteArray &identifier, const BufferAdaptor &bufferAdaptor, Akonadi2::Storage::Transaction &transaction)
 {
-    const auto uid = type.getProperty("uid");
+    const auto uid = bufferAdaptor.getProperty("uid");
     if (uid.isValid()) {
-        Index uidIndex("mail.index.uid", transaction);
-        uidIndex.add(uid.toByteArray(), type.identifier());
+        Index("mail.index.uid", transaction).add(uid.toByteArray(), identifier);
     }
 }
 
diff --git a/common/domain/mail.h b/common/domain/mail.h
index b58ce44..38f1d03 100644
--- a/common/domain/mail.h
+++ b/common/domain/mail.h
@@ -51,7 +51,7 @@ public:
      * An empty result set indicates that a full scan is required.
      */
     static ResultSet queryIndexes(const Akonadi2::Query &query, const QByteArray &resourceInstanceIdentifier, QSet<QByteArray> &appliedFilters, Akonadi2::Storage::Transaction &transaction);
-    static void index(const Mail &type, Akonadi2::Storage::Transaction &transaction);
+    static void index(const QByteArray &identifier, const BufferAdaptor &bufferAdaptor, Akonadi2::Storage::Transaction &transaction);
     static QSharedPointer<ReadPropertyMapper<Buffer> > initializeReadPropertyMapper();
     static QSharedPointer<WritePropertyMapper<BufferBuilder> > initializeWritePropertyMapper();
 };
diff --git a/common/genericresource.cpp b/common/genericresource.cpp
index dcae43d..ec68f33 100644
--- a/common/genericresource.cpp
+++ b/common/genericresource.cpp
@@ -285,10 +285,12 @@ GenericResource::~GenericResource()
     delete mSourceChangeReplay;
 }
 
-// void GenericResource::revisionChanged()
-// {
-//     //TODO replay revision
-// }
+void GenericResource::removeFromDisk(const QByteArray &instanceIdentifier)
+{
+    Akonadi2::Storage(Akonadi2::storageLocation(), instanceIdentifier, Akonadi2::Storage::ReadWrite).removeFromDisk();
+    Akonadi2::Storage(Akonadi2::storageLocation(), instanceIdentifier + ".userqueue", Akonadi2::Storage::ReadWrite).removeFromDisk();
+    Akonadi2::Storage(Akonadi2::storageLocation(), instanceIdentifier + ".synchronizerqueue", Akonadi2::Storage::ReadWrite).removeFromDisk();
+}
 
 void GenericResource::onProcessorError(int errorCode, const QString &errorMessage)
 {
diff --git a/common/genericresource.h b/common/genericresource.h
index cfc6653..33de0e7 100644
--- a/common/genericresource.h
+++ b/common/genericresource.h
@@ -48,6 +48,7 @@ public:
 
     int error() const;
 
+    static void removeFromDisk(const QByteArray &instanceIdentifier);
 private Q_SLOTS:
     void updateLowerBoundRevision();
 
diff --git a/common/pipeline.cpp b/common/pipeline.cpp
index 15d2401..de63288 100644
--- a/common/pipeline.cpp
+++ b/common/pipeline.cpp
@@ -1,5 +1,6 @@
 /*
  * Copyright (C) 2014 Aaron Seigo <aseigo@kde.org>
+ * Copyright (C) 2015 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
@@ -41,19 +42,13 @@ class Pipeline::Private
 {
 public:
     Private(const QString &resourceName)
-        : storage(Akonadi2::storageLocation(), resourceName, Storage::ReadWrite),
-          stepScheduled(false)
+        : storage(Akonadi2::storageLocation(), resourceName, Storage::ReadWrite)
     {
     }
 
     Storage storage;
     Storage::Transaction transaction;
-    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;
+    QHash<QString, QVector<Preprocessor *> > processors;
     QHash<QString, DomainTypeAdaptorFactoryInterface::Ptr> adaptorFactory;
 };
 
@@ -68,21 +63,9 @@ Pipeline::~Pipeline()
     delete d;
 }
 
-void Pipeline::setPreprocessors(const QString &entityType, Type pipelineType, const QVector<Preprocessor *> &preprocessors)
+void Pipeline::setPreprocessors(const QString &entityType, const QVector<Preprocessor *> &processors)
 {
-    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;
-    };
+    d->processors[entityType] = processors;
 }
 
 void Pipeline::setAdaptorFactory(const QString &entityType, DomainTypeAdaptorFactoryInterface::Ptr factory)
@@ -92,6 +75,11 @@ void Pipeline::setAdaptorFactory(const QString &entityType, DomainTypeAdaptorFac
 
 void Pipeline::startTransaction()
 {
+    //TODO call for all types
+    //But avoid doing it during cleanup
+    // for (auto processor : d->processors[bufferType]) {
+    //     processor->startBatch();
+    // }
     if (d->transaction) {
         return;
     }
@@ -100,7 +88,13 @@ void Pipeline::startTransaction()
 
 void Pipeline::commit()
 {
+    //TODO call for all types
+    //But avoid doing it during cleanup
+    // for (auto processor : d->processors[bufferType]) {
+    //     processor->finalize();
+    // }
     const auto revision = Akonadi2::Storage::maxRevision(d->transaction);
+    Trace() << "Committing " << revision;
     if (d->transaction) {
         d->transaction.commit();
     }
@@ -118,14 +112,6 @@ Storage &Pipeline::storage() const
     return d->storage;
 }
 
-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();
-}
-
 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()),
@@ -181,15 +167,25 @@ KAsync::Job<qint64> Pipeline::newEntity(void const *command, size_t size)
 
     storeNewRevision(newRevision, fbb, bufferType, key);
 
-    Log() << "Pipeline: wrote entity: " << key << newRevision << bufferType;
+    auto adaptorFactory = d->adaptorFactory.value(bufferType);
+    if (!adaptorFactory) {
+        Warning() << "no adaptor factory for type " << bufferType;
+        return KAsync::error<qint64>(0);
+    }
 
-    return KAsync::start<qint64>([this, key, bufferType, newRevision](KAsync::Future<qint64> &future) {
-        PipelineState state(this, NewPipeline, Akonadi2::Storage::assembleKey(key, newRevision), d->newPipeline[bufferType], newRevision, [&future, newRevision]() {
-            future.setValue(newRevision);
-            future.setFinished();
-        }, bufferType);
-        d->activePipelines << state;
-        state.step();
+    Log() << "Pipeline: wrote entity: " << key << newRevision << bufferType;
+    d->transaction.openDatabase(bufferType + ".main").scan(Akonadi2::Storage::assembleKey(key, newRevision), [this, bufferType, newRevision, adaptorFactory, key](const QByteArray &, const QByteArray &value) -> bool {
+        auto entity = Akonadi2::GetEntity(value);
+        auto adaptor = adaptorFactory->createAdaptor(*entity);
+        for (auto processor : d->processors[bufferType]) {
+            processor->newEntity(key, newRevision, *adaptor, d->transaction);
+        }
+        return false;
+    }, [this](const Akonadi2::Storage::Error &error) {
+        ErrorMsg() << "Failed to find value in pipeline: " << error.message;
+    });
+    return KAsync::start<qint64>([newRevision](){
+        return newRevision;
     });
 }
 
@@ -287,14 +283,18 @@ KAsync::Job<qint64> Pipeline::modifiedEntity(void const *command, size_t size)
 
     storeNewRevision(newRevision, fbb, bufferType, key);
     Log() << "Pipeline: modified entity: " << key << newRevision << bufferType;
-
-    return KAsync::start<qint64>([this, key, bufferType, newRevision](KAsync::Future<qint64> &future) {
-        PipelineState state(this, ModifiedPipeline, Akonadi2::Storage::assembleKey(key, newRevision), d->modifiedPipeline[bufferType], newRevision, [&future, newRevision]() {
-            future.setValue(newRevision);
-            future.setFinished();
-        }, bufferType);
-        d->activePipelines << state;
-        state.step();
+    d->transaction.openDatabase(bufferType + ".main").scan(Akonadi2::Storage::assembleKey(key, newRevision), [this, bufferType, newRevision, adaptorFactory, current, key](const QByteArray &, const QByteArray &value) -> bool {
+        auto entity = Akonadi2::GetEntity(value);
+        auto newEntity = adaptorFactory->createAdaptor(*entity);
+        for (auto processor : d->processors[bufferType]) {
+            processor->modifiedEntity(key, newRevision, *current, *newEntity, d->transaction);
+        }
+        return false;
+    }, [this](const Akonadi2::Storage::Error &error) {
+        ErrorMsg() << "Failed to find value in pipeline: " << error.message;
+    });
+    return KAsync::start<qint64>([newRevision](){
+        return newRevision;
     });
 }
 
@@ -331,13 +331,24 @@ KAsync::Job<qint64> Pipeline::deletedEntity(void const *command, size_t size)
     storeNewRevision(newRevision, fbb, bufferType, key);
     Log() << "Pipeline: deleted entity: "<< newRevision;
 
-    return KAsync::start<qint64>([this, key, bufferType, newRevision](KAsync::Future<qint64> &future) {
-        PipelineState state(this, DeletedPipeline, key, d->deletedPipeline[bufferType], newRevision, [&future, newRevision](){
-            future.setValue(newRevision);
-            future.setFinished();
-        }, bufferType);
-        d->activePipelines << state;
-        state.step();
+    auto adaptorFactory = d->adaptorFactory.value(bufferType);
+    if (!adaptorFactory) {
+        Warning() << "no adaptor factory for type " << bufferType;
+        return KAsync::error<qint64>(0);
+    }
+
+    // d->transaction.openDatabase(bufferType + ".main").scan(key, [this, bufferType, newRevision, adaptorFactory](const QByteArray &, const QByteArray &value) -> bool {
+    //     auto entity = Akonadi2::GetEntity(value);
+    //     auto newEntity = adaptorFactory->createAdaptor(*entity);
+        for (auto processor : d->processors[bufferType]) {
+            processor->deletedEntity(key, newRevision, Akonadi2::ApplicationDomain::BufferAdaptor(), d->transaction);
+        }
+    //     return false;
+    // }, [this](const Akonadi2::Storage::Error &error) {
+    //     ErrorMsg() << "Failed to find value in pipeline: " << error.message;
+    // });
+    return KAsync::start<qint64>([newRevision](){
+        return newRevision;
     });
 }
 
@@ -372,164 +383,6 @@ qint64 Pipeline::cleanedUpRevision()
     return Akonadi2::Storage::cleanedUpRevision(d->transaction);
 }
 
-void Pipeline::pipelineStepped(const PipelineState &state)
-{
-    scheduleStep();
-}
-
-void Pipeline::scheduleStep()
-{
-    if (!d->stepScheduled) {
-        d->stepScheduled = true;
-        QMetaObject::invokeMethod(this, "stepPipelines", Qt::QueuedConnection);
-    }
-}
-
-void Pipeline::stepPipelines()
-{
-    d->stepScheduled = false;
-    for (PipelineState &state: d->activePipelines) {
-        if (state.isIdle()) {
-            state.step();
-        }
-    }
-}
-
-void Pipeline::pipelineCompleted(PipelineState state)
-{
-    //TODO finalize the datastore, inform clients of the new rev
-    const int index = d->activePipelines.indexOf(state);
-    if (index > -1) {
-        d->activePipelines.remove(index);
-    }
-    state.callback();
-
-    scheduleStep();
-    if (d->activePipelines.isEmpty()) {
-        emit pipelinesDrained();
-    }
-}
-
-
-class PipelineState::Private : public QSharedData
-{
-public:
-    Private(Pipeline *p, Pipeline::Type t, const QByteArray &k, QVector<Preprocessor *> filters, const std::function<void()> &c, qint64 r, const QByteArray &b)
-        : pipeline(p),
-          type(t),
-          key(k),
-          filterIt(filters),
-          idle(true),
-          callback(c),
-          revision(r),
-          bufferType(b)
-    {}
-
-    Private()
-        : pipeline(0),
-          filterIt(QVector<Preprocessor *>()),
-          idle(true),
-          revision(-1)
-    {}
-
-    Pipeline *pipeline;
-    Pipeline::Type type;
-    QByteArray key;
-    QVectorIterator<Preprocessor *> filterIt;
-    bool idle;
-    std::function<void()> callback;
-    qint64 revision;
-    QByteArray bufferType;
-};
-
-PipelineState::PipelineState()
-    : d(new Private())
-{
-
-}
-
-PipelineState::PipelineState(Pipeline *pipeline, Pipeline::Type type, const QByteArray &key, const QVector<Preprocessor *> &filters, qint64 revision, const std::function<void()> &callback, const QByteArray &bufferType)
-    : d(new Private(pipeline, type, key, filters, callback, revision, bufferType))
-{
-}
-
-PipelineState::PipelineState(const PipelineState &other)
-    : d(other.d)
-{
-}
-
-PipelineState::~PipelineState()
-{
-}
-
-PipelineState &PipelineState::operator=(const PipelineState &rhs)
-{
-    d = rhs.d;
-    return *this;
-}
-
-bool PipelineState::operator==(const PipelineState &rhs)
-{
-    return d == rhs.d;
-}
-
-bool PipelineState::isIdle() const
-{
-    return d->idle;
-}
-
-QByteArray PipelineState::key() const
-{
-    return d->key;
-}
-
-Pipeline::Type PipelineState::type() const
-{
-    return d->type;
-}
-
-qint64 PipelineState::revision() const
-{
-    return d->revision;
-}
-
-QByteArray PipelineState::bufferType() const
-{
-    return d->bufferType;
-}
-
-void PipelineState::step()
-{
-    if (!d->pipeline) {
-        Q_ASSERT(false);
-        return;
-    }
-
-    d->idle = false;
-    if (d->filterIt.hasNext()) {
-        //TODO skip step if already processed
-        auto preprocessor = d->filterIt.next();
-        preprocessor->process(*this, d->pipeline->transaction());
-    } else {
-        //This object becomes invalid after this call
-        d->pipeline->pipelineCompleted(*this);
-    }
-}
-
-void PipelineState::processingCompleted(Preprocessor *filter)
-{
-    if (d->pipeline && filter == d->filterIt.peekPrevious()) {
-        d->idle = true;
-        d->pipeline->pipelineStepped(*this);
-    }
-}
-
-void  PipelineState::callback()
-{
-    d->callback();
-}
-
-
 Preprocessor::Preprocessor()
     : d(0)
 {
@@ -539,19 +392,12 @@ Preprocessor::~Preprocessor()
 {
 }
 
-void Preprocessor::process(const PipelineState &state, Akonadi2::Storage::Transaction &transaction)
-{
-    processingCompleted(state);
-}
-
-void Preprocessor::processingCompleted(PipelineState state)
+void Preprocessor::startBatch()
 {
-    state.processingCompleted(this);
 }
 
-QString Preprocessor::id() const
+void Preprocessor::finalize()
 {
-    return QLatin1String("unknown processor");
 }
 
 } // namespace Akonadi2
diff --git a/common/pipeline.h b/common/pipeline.h
index c8d9ddc..f11d880 100644
--- a/common/pipeline.h
+++ b/common/pipeline.h
@@ -35,7 +35,6 @@
 namespace Akonadi2
 {
 
-class PipelineState;
 class Preprocessor;
 
 class AKONADI2COMMON_EXPORT Pipeline : public QObject
@@ -43,19 +42,16 @@ class AKONADI2COMMON_EXPORT Pipeline : public QObject
     Q_OBJECT
 
 public:
-    enum Type { NullPipeline, NewPipeline, ModifiedPipeline, DeletedPipeline };
-
     Pipeline(const QString &storagePath, QObject *parent = 0);
     ~Pipeline();
 
     Storage &storage() const;
 
-    void setPreprocessors(const QString &entityType, Type pipelineType, const QVector<Preprocessor *> &preprocessors);
+    void setPreprocessors(const QString &entityType, const QVector<Preprocessor *> &preprocessors);
     void startTransaction();
     void commit();
     Storage::Transaction &transaction();
 
-    void null();
     void setAdaptorFactory(const QString &entityType, DomainTypeAdaptorFactoryInterface::Ptr factory);
 
     KAsync::Job<qint64> newEntity(void const *command, size_t size);
@@ -75,104 +71,30 @@ public:
 
 Q_SIGNALS:
     void revisionUpdated(qint64);
-    void pipelinesDrained();
-
-private Q_SLOTS:
-    void stepPipelines();
 
 private:
-    void pipelineStepped(const PipelineState &state);
-    //Don't use a reference here (it would invalidate itself)
-    void pipelineCompleted(PipelineState state);
-    void scheduleStep();
     void storeNewRevision(qint64 newRevision, const flatbuffers::FlatBufferBuilder &fbb, const QByteArray &bufferType, const QByteArray &uid);
 
-    friend class PipelineState;
-
     class Private;
     Private * const d;
 };
 
-class AKONADI2COMMON_EXPORT PipelineState
-{
-public:
-    PipelineState();
-    PipelineState(Pipeline *pipeline, Pipeline::Type type, const QByteArray &key, const QVector<Preprocessor *> &filters, qint64 revision, const std::function<void()> &callback, const QByteArray &bufferType);
-    PipelineState(const PipelineState &other);
-    ~PipelineState();
-
-    PipelineState &operator=(const PipelineState &rhs);
-    bool operator==(const PipelineState &rhs);
-
-    bool isIdle() const;
-    QByteArray key() const;
-    Pipeline::Type type() const;
-    qint64 revision() const;
-    QByteArray bufferType() const;
-
-    void step();
-    void processingCompleted(Preprocessor *filter);
-
-    void callback();
-
-private:
-    class Private;
-    QExplicitlySharedDataPointer<Private> d;
-};
-
 class AKONADI2COMMON_EXPORT Preprocessor
 {
 public:
     Preprocessor();
     virtual ~Preprocessor();
 
-    virtual void process(const PipelineState &state, Akonadi2::Storage::Transaction &transaction);
-    //TODO to record progress
-    virtual QString id() const;
-
-protected:
-    void processingCompleted(PipelineState state);
+    virtual void startBatch();
+    virtual void newEntity(const QByteArray &key, qint64 revision, const Akonadi2::ApplicationDomain::BufferAdaptor &newEntity, Akonadi2::Storage::Transaction &transaction) = 0;
+    virtual void modifiedEntity(const QByteArray &key, qint64 revision, const Akonadi2::ApplicationDomain::BufferAdaptor &oldEntity, const Akonadi2::ApplicationDomain::BufferAdaptor &newEntity, Akonadi2::Storage::Transaction &transaction) = 0;
+    virtual void deletedEntity(const QByteArray &key, qint64 revision, const Akonadi2::ApplicationDomain::BufferAdaptor &oldEntity, Akonadi2::Storage::Transaction &transaction) = 0;
+    virtual void finalize();
 
 private:
     class Private;
     Private * const d;
 };
 
-/**
- * A simple processor that takes a single function
- */
-class AKONADI2COMMON_EXPORT SimpleProcessor : public Akonadi2::Preprocessor
-{
-public:
-    SimpleProcessor(const QString &id, const std::function<void(const Akonadi2::PipelineState &state, const Akonadi2::Entity &e, Akonadi2::Storage::Transaction &transaction)> &f)
-        : Akonadi2::Preprocessor(),
-        mFunction(f),
-        mId(id)
-    {
-    }
-
-    void process(const PipelineState &state, Akonadi2::Storage::Transaction &transaction) Q_DECL_OVERRIDE
-    {
-        transaction.openDatabase(state.bufferType() + ".main").scan(state.key(), [this, &state, &transaction](const QByteArray &key, const QByteArray &value) -> bool {
-            auto entity = Akonadi2::GetEntity(value);
-            mFunction(state, *entity, transaction);
-            processingCompleted(state);
-            return false;
-        }, [this, state](const Akonadi2::Storage::Error &error) {
-            ErrorMsg() << "Failed to find value in pipeline: " << error.message;
-            processingCompleted(state);
-        });
-    }
-
-    QString id() const Q_DECL_OVERRIDE
-    {
-        return mId;
-    }
-
-protected:
-    std::function<void(const Akonadi2::PipelineState &state, const Akonadi2::Entity &e, Akonadi2::Storage::Transaction &transaction)> mFunction;
-    QString mId;
-};
-
 } // namespace Akonadi2
 
diff --git a/examples/dummyresource/resourcefactory.cpp b/examples/dummyresource/resourcefactory.cpp
index 8e6bd42..0a2e90b 100644
--- a/examples/dummyresource/resourcefactory.cpp
+++ b/examples/dummyresource/resourcefactory.cpp
@@ -48,31 +48,76 @@ static void index(const QByteArray &index, const QVariant &value, const QByteArr
     }
 }
 
-DummyResource::DummyResource(const QByteArray &instanceIdentifier, const QSharedPointer<Akonadi2::Pipeline> &pipeline)
-    : Akonadi2::GenericResource(instanceIdentifier, pipeline)
-{
-    auto eventFactory = QSharedPointer<DummyEventAdaptorFactory>::create();
-    const auto resourceIdentifier = mResourceInstanceIdentifier;
+/**
+ * Index types:
+ * * uid - property
+ * 
+ * * Property can be:
+ *     * fixed value like uid
+ *     * fixed value where we want to do smaller/greater-than comparisons. (like start date)
+ *     * range indexes like what date range an event affects.
+ *     * group indexes like tree hierarchies as nested sets
+ */
+template<typename DomainType>
+class IndexUpdater : public Akonadi2::Preprocessor {
+public:
+    IndexUpdater(const QByteArray &index, const QByteArray &type)
+        :mIndexIdentifier(index),
+        mBufferType(type)
+    {
 
-    auto eventIndexer = new Akonadi2::SimpleProcessor("eventIndexer", [eventFactory, resourceIdentifier](const Akonadi2::PipelineState &state, const Akonadi2::Entity &entity, Akonadi2::Storage::Transaction &transaction) {
-        Akonadi2::ApplicationDomain::Event event(resourceIdentifier, Akonadi2::Storage::uidFromKey(state.key()), -1, eventFactory->createAdaptor(entity));
-        Akonadi2::ApplicationDomain::TypeImplementation<Akonadi2::ApplicationDomain::Event>::index(event, transaction);
-        index("event.index.rid", event.getProperty("remoteId"), event.identifier(), transaction);
-    });
+    }
+
+    void newEntity(const QByteArray &uid, qint64 revision, const Akonadi2::ApplicationDomain::BufferAdaptor &newEntity, Akonadi2::Storage::Transaction &transaction) Q_DECL_OVERRIDE
+    {
+        Akonadi2::ApplicationDomain::TypeImplementation<DomainType>::index(uid, newEntity, transaction);
+        add(newEntity.getProperty("remoteId"), uid, transaction);
+    }
 
-    mPipeline->setPreprocessors(ENTITY_TYPE_EVENT, Akonadi2::Pipeline::NewPipeline, QVector<Akonadi2::Preprocessor*>() << eventIndexer);
-    mPipeline->setAdaptorFactory(ENTITY_TYPE_EVENT, eventFactory);
-    //TODO cleanup indexes during removal
+    void modifiedEntity(const QByteArray &key, qint64 revision, const Akonadi2::ApplicationDomain::BufferAdaptor &oldEntity, const Akonadi2::ApplicationDomain::BufferAdaptor &newEntity, Akonadi2::Storage::Transaction &transaction) Q_DECL_OVERRIDE
+    {
+    }
 
+    void deletedEntity(const QByteArray &key, qint64 revision, const Akonadi2::ApplicationDomain::BufferAdaptor &oldEntity, Akonadi2::Storage::Transaction &transaction) Q_DECL_OVERRIDE
     {
-        auto mailFactory = QSharedPointer<DummyMailAdaptorFactory>::create();
-        auto mailIndexer = new Akonadi2::SimpleProcessor("mailIndexer", [mailFactory, resourceIdentifier](const Akonadi2::PipelineState &state, const Akonadi2::Entity &entity, Akonadi2::Storage::Transaction &transaction) {
-            Akonadi2::ApplicationDomain::Mail mail(resourceIdentifier, Akonadi2::Storage::uidFromKey(state.key()), -1, mailFactory->createAdaptor(entity));
-            Akonadi2::ApplicationDomain::TypeImplementation<Akonadi2::ApplicationDomain::Mail>::index(mail, transaction);
-            index("mail.index.rid", mail.getProperty("remoteId"), mail.identifier(), transaction);
-        });
+    }
+private:
+    void add(const QVariant &value, const QByteArray &uid, Akonadi2::Storage::Transaction &transaction)
+    {
+        if (value.isValid()) {
+            Index(mIndexIdentifier, transaction).add(value.toByteArray(), uid);
+        }
+    }
+
+    void remove(const QByteArray &uid, Akonadi2::Storage::Transaction &transaction)
+    {
+        //Knowning the indexed value would probably help removing the uid efficiently. Otherwise we have to execute a full scan.
+        // Index(mIndexIdentifier, transaction).remove(uid);
+    }
+
+    void modify(Akonadi2::Storage::Transaction &transaction)
+    {
+        //Knowning the indexed value would probably help removing the uid efficiently. Otherwise we have to execute a full scan.
+        // Index(mIndexIdentifier, transaction).remove(uid);
+    }
 
-        mPipeline->setPreprocessors(ENTITY_TYPE_MAIL, Akonadi2::Pipeline::NewPipeline, QVector<Akonadi2::Preprocessor*>() << mailIndexer);
+    QByteArray mIndexIdentifier;
+    QByteArray mBufferType;
+};
+
+DummyResource::DummyResource(const QByteArray &instanceIdentifier, const QSharedPointer<Akonadi2::Pipeline> &pipeline)
+    : Akonadi2::GenericResource(instanceIdentifier, pipeline)
+{
+    {
+        auto eventFactory = QSharedPointer<DummyEventAdaptorFactory>::create();
+        auto eventIndexer = new IndexUpdater<Akonadi2::ApplicationDomain::Event>("event.index.rid", ENTITY_TYPE_EVENT);
+        mPipeline->setPreprocessors(ENTITY_TYPE_EVENT, QVector<Akonadi2::Preprocessor*>() << eventIndexer);
+        mPipeline->setAdaptorFactory(ENTITY_TYPE_EVENT, eventFactory);
+    }
+    {
+        auto mailFactory = QSharedPointer<DummyMailAdaptorFactory>::create();
+        auto mailIndexer = new IndexUpdater<Akonadi2::ApplicationDomain::Mail>("mail.index.rid", ENTITY_TYPE_MAIL);
+        mPipeline->setPreprocessors(ENTITY_TYPE_MAIL, QVector<Akonadi2::Preprocessor*>() << mailIndexer);
         mPipeline->setAdaptorFactory(ENTITY_TYPE_MAIL, mailFactory);
     }
 }
@@ -171,9 +216,7 @@ KAsync::Job<void> DummyResource::synchronizeWithSource()
 
 void DummyResource::removeFromDisk(const QByteArray &instanceIdentifier)
 {
-    Akonadi2::Storage(Akonadi2::storageLocation(), instanceIdentifier, Akonadi2::Storage::ReadWrite).removeFromDisk();
-    Akonadi2::Storage(Akonadi2::storageLocation(), instanceIdentifier + ".userqueue", Akonadi2::Storage::ReadWrite).removeFromDisk();
-    Akonadi2::Storage(Akonadi2::storageLocation(), instanceIdentifier + ".synchronizerqueue", Akonadi2::Storage::ReadWrite).removeFromDisk();
+    GenericResource::removeFromDisk(instanceIdentifier);
     Akonadi2::Storage(Akonadi2::storageLocation(), instanceIdentifier + ".event.index.uid", Akonadi2::Storage::ReadWrite).removeFromDisk();
 }
 
diff --git a/tests/genericresourcebenchmark.cpp b/tests/genericresourcebenchmark.cpp
index b8635d7..fbe0d12 100644
--- a/tests/genericresourcebenchmark.cpp
+++ b/tests/genericresourcebenchmark.cpp
@@ -60,6 +60,25 @@ static QByteArray createEntityBuffer()
     return QByteArray(reinterpret_cast<const char *>(fbb.GetBufferPointer()), fbb.GetSize());
 }
 
+class IndexUpdater : public Akonadi2::Preprocessor {
+public:
+    void newEntity(const QByteArray &uid, qint64 revision, const Akonadi2::ApplicationDomain::BufferAdaptor &newEntity, Akonadi2::Storage::Transaction &transaction) Q_DECL_OVERRIDE
+    {
+        for (int i = 0; i < 10; i++) {
+            Index ridIndex(QString("index.index%1").arg(i).toLatin1(), transaction);
+            ridIndex.add("foo", uid);
+        }
+    }
+
+    void modifiedEntity(const QByteArray &key, qint64 revision, const Akonadi2::ApplicationDomain::BufferAdaptor &oldEntity, const Akonadi2::ApplicationDomain::BufferAdaptor &newEntity, Akonadi2::Storage::Transaction &transaction) Q_DECL_OVERRIDE
+    {
+    }
+
+    void deletedEntity(const QByteArray &key, qint64 revision, const Akonadi2::ApplicationDomain::BufferAdaptor &oldEntity, Akonadi2::Storage::Transaction &transaction) Q_DECL_OVERRIDE
+    {
+    }
+};
+
 /**
  * Benchmark write performance of generic resource implementation including queues and pipeline.
  */
@@ -124,19 +143,9 @@ private Q_SLOTS:
 
         auto eventFactory = QSharedPointer<TestEventAdaptorFactory>::create();
         const QByteArray resourceIdentifier = "org.kde.test.instance1";
-        auto eventIndexer = new Akonadi2::SimpleProcessor("eventIndexer", [eventFactory, resourceIdentifier](const Akonadi2::PipelineState &state, const Akonadi2::Entity &entity, Akonadi2::Storage::Transaction &transaction) {
-            auto adaptor = eventFactory->createAdaptor(entity);
-            Akonadi2::ApplicationDomain::Event event(resourceIdentifier, state.key(), -1, adaptor);
-            Akonadi2::ApplicationDomain::TypeImplementation<Akonadi2::ApplicationDomain::Event>::index(event, transaction);
-
-            //Create a bunch of indexes
-            for (int i = 0; i < 10; i++) {
-                Index ridIndex(QString("index.index%1").arg(i).toLatin1(), transaction);
-                ridIndex.add("foo", event.identifier());
-            }
-        });
-
-        pipeline->setPreprocessors("event", Akonadi2::Pipeline::NewPipeline, QVector<Akonadi2::Preprocessor*>() << eventIndexer);
+        auto indexer = QSharedPointer<IndexUpdater>::create();
+
+        pipeline->setPreprocessors("event", QVector<Akonadi2::Preprocessor*>() << indexer.data());
         pipeline->setAdaptorFactory("event", eventFactory);
 
         TestResource resource("org.kde.test.instance1", pipeline);
diff --git a/tests/genericresourcetest.cpp b/tests/genericresourcetest.cpp
index 6dd4108..141a5f8 100644
--- a/tests/genericresourcetest.cpp
+++ b/tests/genericresourcetest.cpp
@@ -32,9 +32,7 @@ private Q_SLOTS:
 
     void init()
     {
-        removeFromDisk("org.kde.test.instance1");
-        removeFromDisk("org.kde.test.instance1.userqueue");
-        removeFromDisk("org.kde.test.instance1.synchronizerqueue");
+        Akonadi2::GenericResource::removeFromDisk("org.kde.test.instance1");
         Akonadi2::Log::setDebugOutputLevel(Akonadi2::Log::Trace);
     }
 
diff --git a/tests/pipelinetest.cpp b/tests/pipelinetest.cpp
index 7efba13..0b4c13e 100644
--- a/tests/pipelinetest.cpp
+++ b/tests/pipelinetest.cpp
@@ -139,6 +139,34 @@ QByteArray deleteEntityCommand(const QByteArray &uid, qint64 revision)
     return command;
 }
 
+class TestProcessor : public Akonadi2::Preprocessor {
+public:
+    void newEntity(const QByteArray &uid, qint64 revision, const Akonadi2::ApplicationDomain::BufferAdaptor &newEntity, Akonadi2::Storage::Transaction &transaction) Q_DECL_OVERRIDE
+    {
+        newUids << uid;
+        newRevisions << revision;
+    }
+
+    void modifiedEntity(const QByteArray &uid, qint64 revision, const Akonadi2::ApplicationDomain::BufferAdaptor &oldEntity, const Akonadi2::ApplicationDomain::BufferAdaptor &newEntity, Akonadi2::Storage::Transaction &transaction) Q_DECL_OVERRIDE
+    {
+        modifiedUids << uid;
+        modifiedRevisions << revision;
+    }
+
+    void deletedEntity(const QByteArray &uid, qint64 revision, const Akonadi2::ApplicationDomain::BufferAdaptor &oldEntity, Akonadi2::Storage::Transaction &transaction) Q_DECL_OVERRIDE
+    {
+        deletedUids << uid;
+        deletedRevisions << revision;
+    }
+
+    QList<QByteArray> newUids;
+    QList<qint64> newRevisions;
+    QList<QByteArray> modifiedUids;
+    QList<qint64> modifiedRevisions;
+    QList<QByteArray> deletedUids;
+    QList<qint64> deletedRevisions;
+};
+
 /**
  * Test of the pipeline implementation to ensure new revisions are created correctly in the database.
  */
@@ -251,6 +279,53 @@ private Q_SLOTS:
         //And all revisions are gone
         QCOMPARE(getKeys("org.kde.pipelinetest.instance1", "event.main").size(), 0);
     }
+
+    void testPreprocessor()
+    {
+        flatbuffers::FlatBufferBuilder entityFbb;
+
+        TestProcessor testProcessor;
+
+        Akonadi2::Pipeline pipeline("org.kde.pipelinetest.instance1");
+        pipeline.setPreprocessors("event", QVector<Akonadi2::Preprocessor*>() << &testProcessor);
+        pipeline.startTransaction();
+        pipeline.setAdaptorFactory("event", QSharedPointer<TestEventAdaptorFactory>::create());
+
+        //Actual test
+        {
+            auto command = createEntityCommand(createEvent(entityFbb));
+            pipeline.newEntity(command.constData(), command.size());
+            QCOMPARE(testProcessor.newUids.size(), 1);
+            QCOMPARE(testProcessor.newRevisions.size(), 1);
+            //Key doesn't contain revision and is just the uid
+            QCOMPARE(testProcessor.newUids.at(0), Akonadi2::Storage::uidFromKey(testProcessor.newUids.at(0)));
+        }
+        pipeline.commit();
+        entityFbb.Clear();
+        pipeline.startTransaction();
+        auto keys = getKeys("org.kde.pipelinetest.instance1", "event.main");
+        QCOMPARE(keys.size(), 1);
+        const auto uid = Akonadi2::Storage::uidFromKey(keys.first());
+        {
+            auto modifyCommand = modifyEntityCommand(createEvent(entityFbb, "summary2"), uid, 1);
+            pipeline.modifiedEntity(modifyCommand.constData(), modifyCommand.size());
+            QCOMPARE(testProcessor.modifiedUids.size(), 1);
+            QCOMPARE(testProcessor.modifiedRevisions.size(), 1);
+            //Key doesn't contain revision and is just the uid
+            QCOMPARE(testProcessor.modifiedUids.at(0), Akonadi2::Storage::uidFromKey(testProcessor.modifiedUids.at(0)));
+        }
+        pipeline.commit();
+        entityFbb.Clear();
+        pipeline.startTransaction();
+        {
+            auto deleteCommand = deleteEntityCommand(uid, 1);
+            pipeline.deletedEntity(deleteCommand.constData(), deleteCommand.size());
+            QCOMPARE(testProcessor.deletedUids.size(), 1);
+            QCOMPARE(testProcessor.deletedUids.size(), 1);
+            //Key doesn't contain revision and is just the uid
+            QCOMPARE(testProcessor.deletedUids.at(0), Akonadi2::Storage::uidFromKey(testProcessor.deletedUids.at(0)));
+        }
+    }
 };
 
 QTEST_MAIN(PipelineTest)