a basically-working Pipeline implementation

still a skeleton rather than a full body with flesh and blood, but
it is getting there!

4 files changed, 268 insertions, 28 deletions

diff --git a/common/pipeline.cpp b/common/pipeline.cpp
index 41def7c..5606c30 100644
--- a/common/pipeline.cpp
+++ b/common/pipeline.cpp
@@ -20,7 +20,9 @@
 
 #include "pipeline.h"
 
+#include <QByteArray>
 #include <QStandardPaths>
+#include <QVector>
 
 namespace Akonadi2
 {
@@ -28,47 +30,214 @@ namespace Akonadi2
 class Pipeline::Private
 {
 public:
-    Private(const QString &storageName)
-        : storage(QStandardPaths::writableLocation(QStandardPaths::QStandardPaths::GenericDataLocation) + "/akonadi2", storageName, Akonadi2::Storage::ReadWrite)
+    Private(const QString &resourceName)
+        : storage(QStandardPaths::writableLocation(QStandardPaths::GenericDataLocation) + "/akonadi2/storage", resourceName),
+          stepScheduled(false)
     {
-
     }
 
-    Akonadi2::Storage storage;
+    Storage storage;
+    QVector<PipelineFilter *> nullPipeline;
+    QVector<PipelineFilter *> newPipeline;
+    QVector<PipelineFilter *> modifiedPipeline;
+    QVector<PipelineFilter *> deletedPipeline;
+    QVector<PipelineState> activePipelines;
+    bool stepScheduled;
 };
 
-Pipeline::Pipeline(const QString &storageName)
-    : d(new Private(storageName))
+Pipeline::Pipeline(const QString &resourceName, QObject *parent)
+    : QObject(parent),
+      d(new Private(resourceName))
 {
 }
 
 Pipeline::~Pipeline()
 {
+    delete d;
 }
 
-Storage &Pipeline::storage()
+Storage &Pipeline::storage() const
 {
     return d->storage;
 }
 
-void Pipeline::null(uint messageId, const char *key, size_t keySize, const char *buffer, size_t bufferSize)
+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::newEntity(const QByteArray &key, flatbuffers::FlatBufferBuilder &entity)
+{
+    PipelineState state(this, NewPipeline, key, d->newPipeline);
+    d->activePipelines << state;
+    state.step();
+}
+
+void Pipeline::modifiedEntity(const QByteArray &key, flatbuffers::FlatBufferBuilder &entityDelta)
+{
+    PipelineState state(this, ModifiedPipeline, key, d->modifiedPipeline);
+    d->activePipelines << state;
+    state.step();
+}
+
+void Pipeline::deletedEntity(const QByteArray &key)
+{
+    PipelineState state(this, DeletedPipeline, key, d->deletedPipeline);
+    d->activePipelines << state;
+    state.step();
+}
+
+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()
+{
+    for (PipelineState &state: d->activePipelines) {
+        if (state.isIdle()) {
+            state.step();
+        }
+    }
+
+    d->stepScheduled = false;
+}
+
+void Pipeline::pipelineCompleted(const 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);
+    }
+
+    if (state.type() != NullPipeline) {
+        emit revisionUpdated();
+    }
+    scheduleStep();
+}
+
+
+class PipelineState::Private : public QSharedData
+{
+public:
+    Private(Pipeline *p, Pipeline::Type t, const QByteArray &k, QVector<PipelineFilter *> filters)
+        : pipeline(p),
+          type(t),
+          key(k),
+          filterIt(filters),
+          idle(true)
+    {}
+
+    Private()
+        : pipeline(0),
+          filterIt(QVector<PipelineFilter *>()),
+          idle(true)
+    {}
+
+    Pipeline *pipeline;
+    Pipeline::Type type;
+    QByteArray key;
+    QVectorIterator<PipelineFilter *> filterIt;
+    bool idle;
+};
+
+PipelineState::PipelineState()
+    : d(new Private())
+{
+
+}
+
+PipelineState::PipelineState(Pipeline *pipeline, Pipeline::Type type, const QByteArray &key, const QVector<PipelineFilter *> &filters)
+    : d(new Private(pipeline, type, key, filters))
+{
+}
+
+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;
+}
+
+void PipelineState::step()
+{
+    if (!d->pipeline) {
+        return;
+    }
+
+    d->idle = false;
+    if (d->filterIt.hasNext()) {
+        d->filterIt.next()->process(*this);
+    } else {
+        d->pipeline->pipelineCompleted(*this);
+    }
+}
+
+void PipelineState::processingCompleted(PipelineFilter *filter)
+{
+    if (d->pipeline && filter == d->filterIt.peekPrevious()) {
+        d->idle = true;
+        d->pipeline->pipelineStepped(*this);
+    }
+}
+
+PipelineFilter::PipelineFilter()
+    : d(0)
 {
-    d->storage.write(key, keySize, buffer, bufferSize);
 }
 
-void Pipeline::newEntity(uint messageId, const char *key, size_t keySize, const char *buffer, size_t bufferSize)
+PipelineFilter::~PipelineFilter()
 {
-    d->storage.write(key, keySize, buffer, bufferSize);
 }
 
-void Pipeline::modifiedEntity(uint messageId, const char *key, size_t keySize, const char *buffer, size_t bufferSize)
+void PipelineFilter::process(PipelineState state)
 {
-    d->storage.write(key, keySize, buffer, bufferSize);
+    processingCompleted(state);
 }
 
-void Pipeline::deletedEntity(uint messageId, const char *key, size_t keySize, const char *buffer, size_t bufferSize)
+void PipelineFilter::processingCompleted(PipelineState state)
 {
-    d->storage.write(key, keySize, buffer, bufferSize);
+    state.processingCompleted(this);
 }
 
 } // namespace Akonadi2
diff --git a/common/pipeline.h b/common/pipeline.h
index 635e630..6ad78df 100644
--- a/common/pipeline.h
+++ b/common/pipeline.h
@@ -20,23 +20,89 @@
 
 #pragma once
 
+#include <flatbuffers/flatbuffers.h>
+
+#include <QSharedDataPointer>
+#include <QObject>
+
 #include <akonadi2common_export.h>
 #include <storage.h>
 
 namespace Akonadi2
 {
 
-class AKONADI2COMMON_EXPORT Pipeline
+class PipelineState;
+class PipelineFilter;
+
+class AKONADI2COMMON_EXPORT Pipeline : public QObject
 {
+    Q_OBJECT
+
 public:
-    Pipeline(const QString &storagePath);
+    enum Type { NullPipeline, NewPipeline, ModifiedPipeline, DeletedPipeline };
+
+    Pipeline(const QString &storagePath, QObject *parent = 0);
     ~Pipeline();
 
-    Storage &storage();
-    void null(uint messageId, const char *key, size_t keySize, const char *buffer, size_t bufferSize);
-    void newEntity(uint messageId, const char *key, size_t keySize, const char *buffer, size_t bufferSize);
-    void modifiedEntity(uint messageId, const char *key, size_t keySize, const char *buffer, size_t bufferSize);
-    void deletedEntity(uint messageId, const char *key, size_t keySize, const char *buffer, size_t bufferSize);
+    Storage &storage() const;
+
+    // domain objects needed here
+    void null();
+    void newEntity(const QByteArray &key, flatbuffers::FlatBufferBuilder &entity);
+    void modifiedEntity(const QByteArray &key, flatbuffers::FlatBufferBuilder &entityDelta);
+    void deletedEntity(const QByteArray &key);
+
+Q_SIGNALS:
+    void revisionUpdated();
+
+private Q_SLOTS:
+    void stepPipelines();
+
+private:
+    void pipelineStepped(const PipelineState &state);
+    void pipelineCompleted(const PipelineState &state);
+    void scheduleStep();
+
+    friend class PipelineState;
+
+    class Private;
+    Private * const d;
+};
+
+class AKONADI2COMMON_EXPORT PipelineState
+{
+public:
+    // domain objects?
+    PipelineState();
+    PipelineState(Pipeline *pipeline, Pipeline::Type type, const QByteArray &key, const QVector<PipelineFilter *> &filters);
+    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;
+
+    void step();
+    void processingCompleted(PipelineFilter *filter);
+
+private:
+    class Private;
+    QExplicitlySharedDataPointer<Private> d;
+};
+
+class AKONADI2COMMON_EXPORT PipelineFilter
+{
+public:
+    PipelineFilter();
+    virtual ~PipelineFilter();
+
+    virtual void process(PipelineState state);
+
+protected:
+    void processingCompleted(PipelineState state);
 
 private:
     class Private;
diff --git a/common/resource.cpp b/common/resource.cpp
index 11a03ca..ae28485 100644
--- a/common/resource.cpp
+++ b/common/resource.cpp
@@ -39,12 +39,18 @@ Resource::~Resource()
     //delete d;
 }
 
-void Resource::processCommand(uint messageId, int commandId, const QByteArray &data, uint size, Pipeline *pipeline)
+void Resource::processCommand(int commandId, const QByteArray &data, uint size, Pipeline *pipeline)
 {
+    Q_UNUSED(commandId)
+    Q_UNUSED(data)
+    Q_UNUSED(size)
+    Q_UNUSED(pipeline)
+    pipeline->null();
 }
 
-void Resource::synchronizeWithSource()
+void Resource::synchronizeWithSource(Pipeline *pipeline)
 {
+    pipeline->null();
 }
 
 class ResourceFactory::Private
diff --git a/common/resource.h b/common/resource.h
index 53c0bc1..0f65e1f 100644
--- a/common/resource.h
+++ b/common/resource.h
@@ -18,8 +18,8 @@
  * License along with this library.  If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include <clientapi.h>
 #include <akonadi2common_export.h>
+#include <clientapi.h>
 #include <pipeline.h>
 
 namespace Akonadi2
@@ -32,9 +32,8 @@ public:
     Resource();
     virtual ~Resource();
 
-    //TODO: this will need to be async
-    virtual void processCommand(uint messageId, int commandId, const QByteArray &data, uint size, Pipeline *pipeline);
-    virtual void synchronizeWithSource();
+    virtual void processCommand(int commandId, const QByteArray &data, uint size, Pipeline *pipeline);
+    virtual void synchronizeWithSource(Pipeline *pipeline);
 
 private:
     class Private;