4 files changed, 17 insertions, 21 deletions

diff --git a/common/pipeline.cpp b/common/pipeline.cpp
index 18b6d51..10bae54 100644
--- a/common/pipeline.cpp
+++ b/common/pipeline.cpp
@@ -189,7 +189,6 @@ public:
     Pipeline *pipeline;
     Pipeline::Type type;
     QByteArray key;
-    Akonadi2::Entity *entity;
     QVectorIterator<Preprocessor *> filterIt;
     bool idle;
 };
@@ -240,11 +239,6 @@ Pipeline::Type PipelineState::type() const
     return d->type;
 }
 
-const Akonadi2::Entity &PipelineState::entity() const
-{
-    return *d->entity;
-}
-
 void PipelineState::step()
 {
     if (!d->pipeline) {
@@ -253,7 +247,11 @@ void PipelineState::step()
 
     d->idle = false;
     if (d->filterIt.hasNext()) {
-        d->filterIt.next()->process(*this);
+        d->pipeline->storage().scan(d->key.toStdString(), [this](void *keyValue, int keySize, void *dataValue, int dataSize) -> bool {
+            auto entity = Akonadi2::GetEntity(dataValue);
+            d->filterIt.next()->process(*this, *entity);
+            return false;
+        });
     } else {
         d->pipeline->pipelineCompleted(*this);
     }
@@ -276,7 +274,7 @@ Preprocessor::~Preprocessor()
 {
 }
 
-void Preprocessor::process(PipelineState state)
+void Preprocessor::process(PipelineState state, const Akonadi2::Entity &)
 {
     processingCompleted(state);
 }
diff --git a/common/pipeline.h b/common/pipeline.h
index 6005331..6b847f5 100644
--- a/common/pipeline.h
+++ b/common/pipeline.h
@@ -114,7 +114,6 @@ public:
     bool isIdle() const;
     QByteArray key() const;
     Pipeline::Type type() const;
-    const Akonadi2::Entity &entity() const;
 
     void step();
     void processingCompleted(Preprocessor *filter);
@@ -130,7 +129,7 @@ public:
     Preprocessor();
     virtual ~Preprocessor();
 
-    virtual void process(PipelineState state);
+    virtual void process(PipelineState state, const Akonadi2::Entity &);
 
 protected:
     void processingCompleted(PipelineState state);
diff --git a/dummyresource/resourcefactory.cpp b/dummyresource/resourcefactory.cpp
index da6969a..87a6048 100644
--- a/dummyresource/resourcefactory.cpp
+++ b/dummyresource/resourcefactory.cpp
@@ -38,18 +38,19 @@
 class SimpleProcessor : public Akonadi2::Preprocessor
 {
 public:
-    SimpleProcessor(const std::function<void(const Akonadi2::PipelineState &state)> &f)
+    SimpleProcessor(const std::function<void(const Akonadi2::PipelineState &state, const Akonadi2::Entity &e)> &f)
         : Akonadi2::Preprocessor(),
         mFunction(f)
     {
     }
 
-    void process(const Akonadi2::PipelineState &state) {
-        mFunction(state);
+    void process(const Akonadi2::PipelineState &state, const Akonadi2::Entity &e) {
+        mFunction(state, e);
+        processingCompleted(state);
     }
 
 protected:
-    std::function<void(const Akonadi2::PipelineState &state)> mFunction;
+    std::function<void(const Akonadi2::PipelineState &state, const Akonadi2::Entity &e)> mFunction;
 };
 
 // template <typename DomainType>
@@ -102,13 +103,12 @@ DummyResource::DummyResource()
 
 void DummyResource::configurePipeline(Akonadi2::Pipeline *pipeline)
 {
-    auto factory = QSharedPointer<DummyEventAdaptorFactory>::create();
+    auto eventFactory = QSharedPointer<DummyEventAdaptorFactory>::create();
     //TODO setup preprocessors for each domain type and pipeline type allowing full customization
     //Eventually the order should be self configuring, for now it's hardcoded.
-    auto eventIndexer = new SimpleProcessor([factory](const Akonadi2::PipelineState &state) {
-        auto adaptor = factory->createAdaptor(state.entity());
-        //Here we can plug in generic preprocessors
-        qDebug() << adaptor->getProperty("summary").toString();
+    auto eventIndexer = new SimpleProcessor([eventFactory](const Akonadi2::PipelineState &state, const Akonadi2::Entity &entity) {
+        auto adaptor = eventFactory->createAdaptor(entity);
+        qDebug() << "Summary preprocessor: " << adaptor->getProperty("summary").toString();
     });
     pipeline->setPreprocessors<Akonadi2::Domain::Event>(Akonadi2::Pipeline::NewPipeline, QVector<Akonadi2::Preprocessor*>() << eventIndexer);
 }
diff --git a/synchronizer/listener.cpp b/synchronizer/listener.cpp
index 8b5a19a..dc0d9dd 100644
--- a/synchronizer/listener.cpp
+++ b/synchronizer/listener.cpp
@@ -37,12 +37,10 @@ Listener::Listener(const QString &resourceName, QObject *parent)
       m_server(new QLocalServer(this)),
       m_resourceName(resourceName),
       m_resource(0),
-      //TODO move pipeline(s) to resource
       m_pipeline(new Akonadi2::Pipeline(resourceName, parent)),
       m_clientBufferProcessesTimer(new QTimer(this)),
       m_messageId(0)
 {
-    m_resource->configurePipeline(m_pipeline);
     connect(m_pipeline, &Akonadi2::Pipeline::revisionUpdated,
             this, &Listener::refreshRevision);
     connect(m_server, &QLocalServer::newConnection,
@@ -313,6 +311,7 @@ void Listener::loadResource()
         log(QString("\tResource: %1").arg((qlonglong)m_resource));
         //TODO: this doesn't really list all the facades .. fix
         log(QString("\tFacades: %1").arg(Akonadi2::FacadeFactory::instance().getFacade<Akonadi2::Domain::Event>(m_resourceName)->type()));
+        m_resource->configurePipeline(m_pipeline);
     } else {
         log(QString("Failed to load resource %1").arg(m_resourceName));
     }