Made the pipeline an implementation detail of the resource.

This removes one dependency from the Listener and will allow us
to test the Listener better.

11 files changed, 46 insertions, 75 deletions

diff --git a/common/genericresource.cpp b/common/genericresource.cpp
index 99d1aaa..4dd73b3 100644
--- a/common/genericresource.cpp
+++ b/common/genericresource.cpp
@@ -152,26 +152,21 @@ private:
 };
 
 
-GenericResource::GenericResource(const QByteArray &resourceInstanceIdentifier)
+GenericResource::GenericResource(const QByteArray &resourceInstanceIdentifier, const QSharedPointer<Pipeline> &pipeline)
     : Akonadi2::Resource(),
     mUserQueue(QStandardPaths::writableLocation(QStandardPaths::GenericDataLocation) + "/akonadi2/storage", resourceInstanceIdentifier + ".userqueue"),
     mSynchronizerQueue(QStandardPaths::writableLocation(QStandardPaths::GenericDataLocation) + "/akonadi2/storage", resourceInstanceIdentifier + ".synchronizerqueue"),
     mResourceInstanceIdentifier(resourceInstanceIdentifier),
+    mPipeline(pipeline ? pipeline : QSharedPointer<Akonadi2::Pipeline>::create(resourceInstanceIdentifier)),
     mError(0)
 {
+    mProcessor = new Processor(mPipeline.data(), QList<MessageQueue*>() << &mUserQueue << &mSynchronizerQueue);
+    QObject::connect(mProcessor, &Processor::error, [this](int errorCode, const QString &msg) { onProcessorError(errorCode, msg); });
+    QObject::connect(mPipeline.data(), &Pipeline::revisionUpdated, this, &Resource::revisionUpdated);
 }
 
 GenericResource::~GenericResource()
 {
-
-}
-
-void GenericResource::configurePipeline(Akonadi2::Pipeline *pipeline)
-{
-    //TODO figure out lifetime of the processor
-    mProcessor = new Processor(pipeline, QList<MessageQueue*>() << &mUserQueue << &mSynchronizerQueue);
-    QObject::connect(mProcessor, &Processor::error, [this](int errorCode, const QString &msg) { onProcessorError(errorCode, msg); });
-    QObject::connect(pipeline, &Pipeline::revisionUpdated, this, &Resource::revisionUpdated);
 }
 
 void GenericResource::onProcessorError(int errorCode, const QString &errorMessage)
@@ -195,7 +190,7 @@ void GenericResource::enqueueCommand(MessageQueue &mq, int commandId, const QByt
     mq.enqueue(m_fbb.GetBufferPointer(), m_fbb.GetSize());
 }
 
-void GenericResource::processCommand(int commandId, const QByteArray &data, Akonadi2::Pipeline *pipeline)
+void GenericResource::processCommand(int commandId, const QByteArray &data)
 {
     //TODO instead of copying the command including the full entity first into the command queue, we could directly
     //create a new revision, only pushing a handle into the commandqueue with the relevant changeset (for changereplay).
diff --git a/common/genericresource.h b/common/genericresource.h
index e9d5d59..4a285ea 100644
--- a/common/genericresource.h
+++ b/common/genericresource.h
@@ -22,6 +22,7 @@
 #include <akonadi2common_export.h>
 #include <resource.h>
 #include <messagequeue.h>
+#include <flatbuffers/flatbuffers.h>
 
 class Processor;
 
@@ -34,14 +35,13 @@ namespace Akonadi2
 class AKONADI2COMMON_EXPORT GenericResource : public Resource
 {
 public:
-    GenericResource(const QByteArray &resourceInstanceIdentifier);
+    GenericResource(const QByteArray &resourceInstanceIdentifier, const QSharedPointer<Pipeline> &pipeline = QSharedPointer<Pipeline>());
     virtual ~GenericResource();
 
-    virtual void processCommand(int commandId, const QByteArray &data, Pipeline *pipeline) Q_DECL_OVERRIDE;
-    virtual KAsync::Job<void> synchronizeWithSource(Pipeline *pipeline) Q_DECL_OVERRIDE = 0;
+    virtual void processCommand(int commandId, const QByteArray &data) Q_DECL_OVERRIDE;
+    virtual KAsync::Job<void> synchronizeWithSource() Q_DECL_OVERRIDE = 0;
     virtual KAsync::Job<void> processAllMessages() Q_DECL_OVERRIDE;
 
-    virtual void configurePipeline(Pipeline *pipeline) Q_DECL_OVERRIDE;
     int error() const;
 
 protected:
@@ -51,6 +51,7 @@ protected:
     MessageQueue mUserQueue;
     MessageQueue mSynchronizerQueue;
     QByteArray mResourceInstanceIdentifier;
+    QSharedPointer<Pipeline> mPipeline;
 
 private:
     Processor *mProcessor;
diff --git a/common/listener.cpp b/common/listener.cpp
index 2e2e98e..8ec9b3e 100644
--- a/common/listener.cpp
+++ b/common/listener.cpp
@@ -41,7 +41,6 @@ Listener::Listener(const QByteArray &resourceInstanceIdentifier, QObject *parent
       m_resourceName(Akonadi2::Store::resourceName(resourceInstanceIdentifier)),
       m_resourceInstanceIdentifier(resourceInstanceIdentifier),
       m_resource(0),
-      m_pipeline(new Akonadi2::Pipeline(resourceInstanceIdentifier, parent)),
       m_clientBufferProcessesTimer(new QTimer(this)),
       m_messageId(0)
 {
@@ -226,7 +225,7 @@ void Listener::processCommand(int commandId, uint messageId, const QByteArray &c
                 }
                 auto job = KAsync::null<void>();
                 if (buffer->sourceSync()) {
-                    job = m_resource->synchronizeWithSource(m_pipeline);
+                    job = m_resource->synchronizeWithSource();
                 }
                 if (buffer->localSync()) {
                     job = job.then<void>(m_resource->processAllMessages());
@@ -247,7 +246,7 @@ void Listener::processCommand(int commandId, uint messageId, const QByteArray &c
             Log() << "\tCommand id  " << messageId << " of type \"" << Akonadi2::Commands::name(commandId) << "\" from " << client.name;
             loadResource();
             if (m_resource) {
-                m_resource->processCommand(commandId, commandBuffer, m_pipeline);
+                m_resource->processCommand(commandId, commandBuffer);
             }
             break;
         case Akonadi2::Commands::ShutdownCommand:
@@ -261,7 +260,7 @@ void Listener::processCommand(int commandId, uint messageId, const QByteArray &c
                 Log() << QString("\tReceived custom command from %1: ").arg(client.name) << commandId;
                 loadResource();
                 if (m_resource) {
-                    m_resource->processCommand(commandId, commandBuffer, m_pipeline);
+                    m_resource->processCommand(commandId, commandBuffer);
                 }
             } else {
                 Warning() << QString("\tReceived invalid command from %1: ").arg(client.name) << commandId;
@@ -367,7 +366,6 @@ void Listener::loadResource()
         m_resource = resourceFactory->createResource(m_resourceInstanceIdentifier);
         Log() << QString("Resource factory: %1").arg((qlonglong)resourceFactory);
         Log() << QString("\tResource: %1").arg((qlonglong)m_resource);
-        m_resource->configurePipeline(m_pipeline);
         connect(m_resource, &Akonadi2::Resource::revisionUpdated,
             this, &Listener::refreshRevision);
     } else {
diff --git a/common/listener.h b/common/listener.h
index 649c3ed..0d19823 100644
--- a/common/listener.h
+++ b/common/listener.h
@@ -92,7 +92,6 @@ private:
     const QByteArray m_resourceName;
     const QByteArray m_resourceInstanceIdentifier;
     Akonadi2::Resource *m_resource;
-    Akonadi2::Pipeline *m_pipeline;
     QTimer *m_clientBufferProcessesTimer;
     QTimer *m_checkConnectionsTimer;
     int m_messageId;
diff --git a/common/resource.cpp b/common/resource.cpp
index 68a237c..2a86df5 100644
--- a/common/resource.cpp
+++ b/common/resource.cpp
@@ -40,24 +40,15 @@ Resource::~Resource()
     //delete d;
 }
 
-void Resource::configurePipeline(Pipeline *pipeline)
-{
-
-}
-
-void Resource::processCommand(int commandId, const QByteArray &data, Pipeline *pipeline)
+void Resource::processCommand(int commandId, const QByteArray &data)
 {
     Q_UNUSED(commandId)
     Q_UNUSED(data)
-    Q_UNUSED(pipeline)
-    pipeline->null();
 }
 
-KAsync::Job<void> Resource::synchronizeWithSource(Pipeline *pipeline)
+KAsync::Job<void> Resource::synchronizeWithSource()
 {
-    return KAsync::start<void>([pipeline](KAsync::Future<void> &f) {
-        pipeline->null();
-    });
+    return KAsync::null<void>();
 }
 
 KAsync::Job<void> Resource::processAllMessages()
diff --git a/common/resource.h b/common/resource.h
index 9f657f7..a51e12d 100644
--- a/common/resource.h
+++ b/common/resource.h
@@ -20,12 +20,12 @@
 
 #include <akonadi2common_export.h>
 #include <clientapi.h>
-#include <pipeline.h>
 
 #include <Async/Async>
 
 namespace Akonadi2
 {
+class Pipeline;
 
 /**
  * Resource interface
@@ -37,12 +37,10 @@ public:
     Resource();
     virtual ~Resource();
 
-    virtual void processCommand(int commandId, const QByteArray &data, Pipeline *pipeline);
-    virtual KAsync::Job<void> synchronizeWithSource(Pipeline *pipeline);
+    virtual void processCommand(int commandId, const QByteArray &data);
+    virtual KAsync::Job<void> synchronizeWithSource();
     virtual KAsync::Job<void> processAllMessages();
 
-    virtual void configurePipeline(Pipeline *pipeline);
-
 Q_SIGNALS:
     void revisionUpdated(qint64);
 
diff --git a/examples/dummyresource/resourcefactory.cpp b/examples/dummyresource/resourcefactory.cpp
index c7a3eef..e9bf6cd 100644
--- a/examples/dummyresource/resourcefactory.cpp
+++ b/examples/dummyresource/resourcefactory.cpp
@@ -35,12 +35,8 @@
 //This is the resources entity type, and not the domain type
 #define ENTITY_TYPE_EVENT "event"
 
-DummyResource::DummyResource(const QByteArray &instanceIdentifier)
-    : Akonadi2::GenericResource(instanceIdentifier)
-{
-}
-
-void DummyResource::configurePipeline(Akonadi2::Pipeline *pipeline)
+DummyResource::DummyResource(const QByteArray &instanceIdentifier, const QSharedPointer<Akonadi2::Pipeline> &pipeline)
+    : Akonadi2::GenericResource(instanceIdentifier, pipeline)
 {
     auto eventFactory = QSharedPointer<DummyEventAdaptorFactory>::create();
     const auto resourceIdentifier = mResourceInstanceIdentifier;
@@ -57,15 +53,14 @@ void DummyResource::configurePipeline(Akonadi2::Pipeline *pipeline)
         }
     });
 
-    pipeline->setPreprocessors(ENTITY_TYPE_EVENT, Akonadi2::Pipeline::NewPipeline, QVector<Akonadi2::Preprocessor*>() << eventIndexer);
-    pipeline->setAdaptorFactory(ENTITY_TYPE_EVENT, eventFactory);
+    mPipeline->setPreprocessors(ENTITY_TYPE_EVENT, Akonadi2::Pipeline::NewPipeline, QVector<Akonadi2::Preprocessor*>() << eventIndexer);
+    mPipeline->setAdaptorFactory(ENTITY_TYPE_EVENT, eventFactory);
     //TODO cleanup indexes during removal
-    GenericResource::configurePipeline(pipeline);
 }
 
-KAsync::Job<void> DummyResource::synchronizeWithSource(Akonadi2::Pipeline *pipeline)
+KAsync::Job<void> DummyResource::synchronizeWithSource()
 {
-    return KAsync::start<void>([this, pipeline](KAsync::Future<void> &f) {
+    return KAsync::start<void>([this](KAsync::Future<void> &f) {
         //TODO start transaction on index
         Index uidIndex(Akonadi2::Store::storageLocation(), mResourceInstanceIdentifier + ".index.uid", Akonadi2::Storage::ReadOnly);
 
diff --git a/examples/dummyresource/resourcefactory.h b/examples/dummyresource/resourcefactory.h
index f2362bc..4baafa7 100644
--- a/examples/dummyresource/resourcefactory.h
+++ b/examples/dummyresource/resourcefactory.h
@@ -32,9 +32,8 @@
 class DummyResource : public Akonadi2::GenericResource
 {
 public:
-    DummyResource(const QByteArray &instanceIdentifier);
-    KAsync::Job<void> synchronizeWithSource(Akonadi2::Pipeline *pipeline) Q_DECL_OVERRIDE;
-    void configurePipeline(Akonadi2::Pipeline *pipeline) Q_DECL_OVERRIDE;
+    DummyResource(const QByteArray &instanceIdentifier, const QSharedPointer<Akonadi2::Pipeline> &pipeline = QSharedPointer<Akonadi2::Pipeline>());
+    KAsync::Job<void> synchronizeWithSource() Q_DECL_OVERRIDE;
 };
 
 class DummyResourceFactory : public Akonadi2::ResourceFactory
diff --git a/tests/dummyresourcebenchmark.cpp b/tests/dummyresourcebenchmark.cpp
index 7d40779..4c649a9 100644
--- a/tests/dummyresourcebenchmark.cpp
+++ b/tests/dummyresourcebenchmark.cpp
@@ -8,11 +8,13 @@
 #include "commands.h"
 #include "entitybuffer.h"
 #include "synclistresult.h"
+#include "pipeline.h"
 
 #include "event_generated.h"
 #include "entity_generated.h"
 #include "metadata_generated.h"
 #include "createentity_generated.h"
+
 #include <iostream>
 
 static void removeFromDisk(const QString &name)
@@ -94,10 +96,9 @@ private Q_SLOTS:
         time.start();
         int num = 10000;
 
-        Akonadi2::Pipeline pipeline("org.kde.dummy.instance1");
-        QSignalSpy revisionSpy(&pipeline, SIGNAL(revisionUpdated()));
-        DummyResource resource("org.kde.dummy.instance1");
-        resource.configurePipeline(&pipeline);
+        auto pipeline = QSharedPointer<Akonadi2::Pipeline>::create("org.kde.dummy.instance1");
+        QSignalSpy revisionSpy(pipeline.data(), SIGNAL(revisionUpdated()));
+        DummyResource resource("org.kde.dummy.instance1", pipeline);
 
         flatbuffers::FlatBufferBuilder eventFbb;
         eventFbb.Clear();
@@ -133,7 +134,7 @@ private Q_SLOTS:
         const QByteArray command(reinterpret_cast<const char *>(fbb.GetBufferPointer()), fbb.GetSize());
 
         for (int i = 0; i < num; i++) {
-            resource.processCommand(Akonadi2::Commands::CreateEntityCommand, command, &pipeline);
+            resource.processCommand(Akonadi2::Commands::CreateEntityCommand, command);
         }
         auto appendTime = time.elapsed();
 
diff --git a/tests/dummyresourcetest.cpp b/tests/dummyresourcetest.cpp
index 10cd7e3..1a4d6ca 100644
--- a/tests/dummyresourcetest.cpp
+++ b/tests/dummyresourcetest.cpp
@@ -12,6 +12,7 @@
 #include "commands.h"
 #include "entitybuffer.h"
 #include "resourceconfig.h"
+#include "pipeline.h"
 
 static void removeFromDisk(const QString &name)
 {
@@ -133,10 +134,9 @@ private Q_SLOTS:
 
     void testResourceSync()
     {
-        Akonadi2::Pipeline pipeline("org.kde.dummy.instance1");
-        DummyResource resource("org.kde.dummy.instance1");
-        resource.configurePipeline(&pipeline);
-        auto job = resource.synchronizeWithSource(&pipeline);
+        auto pipeline = QSharedPointer<Akonadi2::Pipeline>::create("org.kde.dummy.instance1");
+        DummyResource resource("org.kde.dummy.instance1", pipeline);
+        auto job = resource.synchronizeWithSource();
         //TODO pass in optional timeout?
         auto future = job.exec();
         future.waitForFinished();
diff --git a/tests/genericresourcetest.cpp b/tests/genericresourcetest.cpp
index 0b9a5c1..b6f629a 100644
--- a/tests/genericresourcetest.cpp
+++ b/tests/genericresourcetest.cpp
@@ -14,20 +14,15 @@
 class TestResource : public Akonadi2::GenericResource
 {
 public:
-    TestResource(const QByteArray &instanceIdentifier)
-        : Akonadi2::GenericResource(instanceIdentifier)
+    TestResource(const QByteArray &instanceIdentifier, QSharedPointer<Akonadi2::Pipeline> pipeline)
+        : Akonadi2::GenericResource(instanceIdentifier, pipeline)
     {
     }
 
-    KAsync::Job<void> synchronizeWithSource(Akonadi2::Pipeline *pipeline) Q_DECL_OVERRIDE
+    KAsync::Job<void> synchronizeWithSource() Q_DECL_OVERRIDE
     {
         return KAsync::null<void>();
     }
-
-    void configurePipeline(Akonadi2::Pipeline *pipeline) Q_DECL_OVERRIDE
-    {
-        GenericResource::configurePipeline(pipeline);
-    }
 };
 
 
@@ -89,12 +84,11 @@ private Q_SLOTS:
         }
 
         //Actual test
-        Akonadi2::Pipeline pipeline("org.kde.test.instance1");
-        QSignalSpy revisionSpy(&pipeline, SIGNAL(revisionUpdated(qint64)));
-        TestResource resource("org.kde.test.instance1");
-        resource.configurePipeline(&pipeline);
-        resource.processCommand(Akonadi2::Commands::CreateEntityCommand, command, &pipeline);
-        resource.processCommand(Akonadi2::Commands::CreateEntityCommand, command, &pipeline);
+        auto pipeline = QSharedPointer<Akonadi2::Pipeline>::create("org.kde.test.instance1");
+        QSignalSpy revisionSpy(pipeline.data(), SIGNAL(revisionUpdated(qint64)));
+        TestResource resource("org.kde.test.instance1", pipeline);
+        resource.processCommand(Akonadi2::Commands::CreateEntityCommand, command);
+        resource.processCommand(Akonadi2::Commands::CreateEntityCommand, command);
 
         QVERIFY(revisionSpy.isValid());
         QTRY_COMPARE(revisionSpy.count(), 2);