Move the commandprocessor to a separate file.

1 files changed, 1 insertions, 188 deletions

diff --git a/common/genericresource.cpp b/common/genericresource.cpp
index 7b83957..3aa4fce 100644
--- a/common/genericresource.cpp
+++ b/common/genericresource.cpp
@@ -36,6 +36,7 @@
 #include "bufferutils.h"
 #include "adaptorfactoryregistry.h"
 #include "synchronizer.h"
+#include "commandprocessor.h"
 
 #include <QUuid>
 #include <QDataStream>
@@ -48,194 +49,6 @@ static int sCommitInterval = 10;
 using namespace Sink;
 using namespace Sink::Storage;
 
-/**
- * Drives the pipeline using the output from all command queues
- */
-class CommandProcessor : public QObject
-{
-    Q_OBJECT
-    typedef std::function<KAsync::Job<void>(void const *, size_t)> InspectionFunction;
-    typedef std::function<KAsync::Job<void>(void const *, size_t)> FlushFunction;
-    SINK_DEBUG_AREA("commandprocessor")
-
-public:
-    CommandProcessor(Sink::Pipeline *pipeline, QList<MessageQueue *> commandQueues) : QObject(), mPipeline(pipeline), mCommandQueues(commandQueues), mProcessingLock(false), mLowerBoundRevision(0)
-    {
-        for (auto queue : mCommandQueues) {
-            const bool ret = connect(queue, &MessageQueue::messageReady, this, &CommandProcessor::process);
-            Q_UNUSED(ret);
-        }
-    }
-
-    void setOldestUsedRevision(qint64 revision)
-    {
-        mLowerBoundRevision = revision;
-    }
-
-    void setInspectionCommand(const InspectionFunction &f)
-    {
-        mInspect = f;
-    }
-
-    void setFlushCommand(const FlushFunction &f)
-    {
-        mFlush = f;
-    }
-
-signals:
-    void error(int errorCode, const QString &errorMessage);
-
-private:
-    bool messagesToProcessAvailable()
-    {
-        for (auto queue : mCommandQueues) {
-            if (!queue->isEmpty()) {
-                return true;
-            }
-        }
-        return false;
-    }
-
-private slots:
-    void process()
-    {
-        if (mProcessingLock) {
-            return;
-        }
-        mProcessingLock = true;
-        auto job = processPipeline()
-                       .syncThen<void>([this]() {
-                           mProcessingLock = false;
-                           if (messagesToProcessAvailable()) {
-                               process();
-                           }
-                       })
-                       .exec();
-    }
-
-    KAsync::Job<qint64> processQueuedCommand(const Sink::QueuedCommand *queuedCommand)
-    {
-        SinkTrace() << "Processing command: " << Sink::Commands::name(queuedCommand->commandId());
-        // Throw command into appropriate pipeline
-        switch (queuedCommand->commandId()) {
-            case Sink::Commands::DeleteEntityCommand:
-                return mPipeline->deletedEntity(queuedCommand->command()->Data(), queuedCommand->command()->size());
-            case Sink::Commands::ModifyEntityCommand:
-                return mPipeline->modifiedEntity(queuedCommand->command()->Data(), queuedCommand->command()->size());
-            case Sink::Commands::CreateEntityCommand:
-                return mPipeline->newEntity(queuedCommand->command()->Data(), queuedCommand->command()->size());
-            case Sink::Commands::InspectionCommand:
-                if (mInspect) {
-                    return mInspect(queuedCommand->command()->Data(), queuedCommand->command()->size())
-                        .syncThen<qint64>([]() { return -1; });
-                } else {
-                    return KAsync::error<qint64>(-1, "Missing inspection command.");
-                }
-            case Sink::Commands::FlushCommand:
-                if (mFlush) {
-                    return mFlush(queuedCommand->command()->Data(), queuedCommand->command()->size())
-                        .syncThen<qint64>([]() { return -1; });
-                } else {
-                    return KAsync::error<qint64>(-1, "Missing inspection command.");
-                }
-            default:
-                return KAsync::error<qint64>(-1, "Unhandled command");
-        }
-    }
-
-    KAsync::Job<qint64> processQueuedCommand(const QByteArray &data)
-    {
-        flatbuffers::Verifier verifyer(reinterpret_cast<const uint8_t *>(data.constData()), data.size());
-        if (!Sink::VerifyQueuedCommandBuffer(verifyer)) {
-            SinkWarning() << "invalid buffer";
-            // return KAsync::error<void, qint64>(1, "Invalid Buffer");
-        }
-        auto queuedCommand = Sink::GetQueuedCommand(data.constData());
-        const auto commandId = queuedCommand->commandId();
-        SinkTrace() << "Dequeued Command: " << Sink::Commands::name(commandId);
-        return processQueuedCommand(queuedCommand)
-            .then<qint64, qint64>(
-                [this, commandId](const KAsync::Error &error, qint64 createdRevision) -> KAsync::Job<qint64> {
-                    if (error) {
-                        SinkWarning() << "Error while processing queue command: " << error.errorMessage;
-                        return KAsync::error<qint64>(error);
-                    }
-                    SinkTrace() << "Command pipeline processed: " << Sink::Commands::name(commandId);
-                    return KAsync::value<qint64>(createdRevision);
-                });
-    }
-
-    // Process all messages of this queue
-    KAsync::Job<void> processQueue(MessageQueue *queue)
-    {
-        auto time = QSharedPointer<QTime>::create();
-        return KAsync::syncStart<void>([this]() { mPipeline->startTransaction(); })
-            .then(KAsync::dowhile(
-                [this, queue, time]() -> KAsync::Job<KAsync::ControlFlowFlag> {
-                    return queue->dequeueBatch(sBatchSize,
-                        [this, time](const QByteArray &data) -> KAsync::Job<void> {
-                            time->start();
-                            return processQueuedCommand(data)
-                            .syncThen<void, qint64>([this, time](qint64 createdRevision) {
-                                SinkTrace() << "Created revision " << createdRevision << ". Processing took: " << Log::TraceTime(time->elapsed());
-                            });
-                        })
-                        .then<KAsync::ControlFlowFlag>([queue](const KAsync::Error &error) {
-                                if (error) {
-                                    if (error.errorCode != MessageQueue::ErrorCodes::NoMessageFound) {
-                                        SinkWarning() << "Error while getting message from messagequeue: " << error.errorMessage;
-                                    }
-                                }
-                                if (queue->isEmpty()) {
-                                    return KAsync::value<KAsync::ControlFlowFlag>(KAsync::Break);
-                                } else {
-                                    return KAsync::value<KAsync::ControlFlowFlag>(KAsync::Continue);
-                                }
-                            });
-                }))
-            .syncThen<void>([this](const KAsync::Error &) { mPipeline->commit(); });
-    }
-
-    KAsync::Job<void> processPipeline()
-    {
-        auto time = QSharedPointer<QTime>::create();
-        time->start();
-        mPipeline->cleanupRevisions(mLowerBoundRevision);
-        SinkTrace() << "Cleanup done." << Log::TraceTime(time->elapsed());
-
-        // Go through all message queues
-        if (mCommandQueues.isEmpty()) {
-            return KAsync::null<void>();
-        }
-        auto it = QSharedPointer<QListIterator<MessageQueue *>>::create(mCommandQueues);
-        return KAsync::dowhile(
-            [it, this]() {
-                auto time = QSharedPointer<QTime>::create();
-                time->start();
-
-                auto queue = it->next();
-                return processQueue(queue)
-                    .syncThen<KAsync::ControlFlowFlag>([this, time, it]() {
-                        SinkTrace() << "Queue processed." << Log::TraceTime(time->elapsed());
-                        if (it->hasNext()) {
-                            return KAsync::Continue;
-                        }
-                        return KAsync::Break;
-                    });
-            });
-    }
-
-private:
-    Sink::Pipeline *mPipeline;
-    // Ordered by priority
-    QList<MessageQueue *> mCommandQueues;
-    bool mProcessingLock;
-    // The lowest revision we no longer need
-    qint64 mLowerBoundRevision;
-    InspectionFunction mInspect;
-    FlushFunction mFlush;
-};
-
 GenericResource::GenericResource(const ResourceContext &resourceContext, const QSharedPointer<Pipeline> &pipeline )
     : Sink::Resource(),
       mResourceContext(resourceContext),