5 files changed, 133 insertions, 0 deletions

diff --git a/common/CMakeLists.txt b/common/CMakeLists.txt
index 1a9a812..671d1cd 100644
--- a/common/CMakeLists.txt
+++ b/common/CMakeLists.txt
@@ -30,6 +30,7 @@ set(command_SRCS
     resourceaccess.cpp
     storage_common.cpp
     threadboundary.cpp
+    messagequeue.cpp
     ${storage_SRCS})
 
 add_library(${PROJECT_NAME} SHARED ${command_SRCS})
diff --git a/common/messagequeue.cpp b/common/messagequeue.cpp
new file mode 100644
index 0000000..cf63e57
--- /dev/null
+++ b/common/messagequeue.cpp
@@ -0,0 +1,47 @@
+#include "messagequeue.h"
+#include <QDebug>
+
+MessageQueue::MessageQueue(const QString &storageRoot, const QString &name)
+    : mStorage(storageRoot, name, Akonadi2::Storage::ReadWrite)
+{
+
+}
+
+void MessageQueue::enqueue(void const *msg, size_t size)
+{
+    mStorage.startTransaction(Akonadi2::Storage::ReadWrite);
+    const qint64 revision = mStorage.maxRevision() + 1;
+    const QByteArray key = QString("%1").arg(revision).toUtf8();
+    mStorage.write(key.data(), key.size(), msg, size);
+    mStorage.setMaxRevision(revision);
+    mStorage.commitTransaction();
+}
+
+void MessageQueue::dequeue(const std::function<void(void *ptr, int size, std::function<void(bool success)>)> &resultHandler,
+                           const std::function<void(const Error &error)> &errorHandler)
+{
+    mStorage.scan("", [this, resultHandler](void *keyPtr, int keySize, void *valuePtr, int valueSize) -> bool {
+        const std::string key(static_cast<char*>(keyPtr), keySize);
+        resultHandler(valuePtr, valueSize, [this, key](bool success) {
+            if (success) {
+                mStorage.remove(key.data(), key.size());
+            } else {
+                //TODO re-enqueue?
+            }
+        });
+        return false;
+    });
+}
+
+bool MessageQueue::isEmpty()
+{
+    int count = 0;
+    mStorage.scan("", [&count](void *keyPtr, int keySize, void *valuePtr, int valueSize) -> bool {
+        const QByteArray key(static_cast<char*>(keyPtr), keySize);
+        if (!key.startsWith("__internal")) {
+            count++;
+        }
+    });
+    return count == 0;
+}
+
diff --git a/common/messagequeue.h b/common/messagequeue.h
new file mode 100644
index 0000000..b56b3cd
--- /dev/null
+++ b/common/messagequeue.h
@@ -0,0 +1,34 @@
+#pragma once
+
+#include <string>
+#include <functional>
+#include <QString>
+#include "storage.h"
+
+/**
+ * A persistent FIFO message queue.
+ */
+class MessageQueue {
+public:
+    class Error
+    {
+    public:
+        Error(const std::string &s, int c, const std::string &m)
+            : store(s), message(m), code(c) {}
+        std::string store;
+        std::string message;
+        int code;
+    };
+
+    MessageQueue(const QString &storageRoot, const QString &name);
+
+    void enqueue(void const *msg, size_t size);
+    //Dequeue a message. This will return a new message everytime called.
+    //Call the result handler with a success response to remove the message from the store.
+    //TODO track processing progress to avoid processing the same message with the same preprocessor twice?
+    void dequeue(const std::function<void(void *ptr, int size, std::function<void(bool success)>)> & resultHandler,
+              const std::function<void(const Error &error)> &errorHandler);
+    bool isEmpty();
+private:
+    Akonadi2::Storage mStorage;
+};
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index 12695ff..183b1bf 100644
--- a/tests/CMakeLists.txt
+++ b/tests/CMakeLists.txt
@@ -18,6 +18,7 @@ manual_tests (
     storagetest
     dummyresourcetest
     domainadaptortest
+    messagequeuetest
 )
 
 target_link_libraries(dummyresourcetest akonadi2_resource_dummy)
diff --git a/tests/messagequeuetest.cpp b/tests/messagequeuetest.cpp
new file mode 100644
index 0000000..2ea9f0c
--- /dev/null
+++ b/tests/messagequeuetest.cpp
@@ -0,0 +1,50 @@
+#include <QtTest>
+
+#include <QString>
+#include <QQueue>
+
+#include "clientapi.h"
+#include "storage.h"
+#include "messagequeue.h"
+
+class MessageQueueTest : public QObject
+{
+    Q_OBJECT
+private Q_SLOTS:
+    void initTestCase()
+    {
+        Akonadi2::Storage store(Akonadi2::Store::storageLocation(), "org.kde.dummy.testqueue", Akonadi2::Storage::ReadWrite);
+        store.removeFromDisk();
+    }
+
+    void cleanupTestCase()
+    {
+    }
+
+    void testQueue()
+    {
+        QQueue<QByteArray> values;
+        values << "value1";
+        values << "value2";
+
+        MessageQueue queue(Akonadi2::Store::storageLocation(), "org.kde.dummy.testqueue");
+        for (const QByteArray &value : values) {
+            queue.enqueue(value.data(), value.size());
+        }
+
+        while (!queue.isEmpty()) {
+            const auto expected = values.dequeue();
+            queue.dequeue([&](void *ptr, int size, std::function<void(bool success)> callback) {
+                QCOMPARE(QByteArray(static_cast<char*>(ptr), size), expected);
+                callback(true);
+            }, [](const MessageQueue::Error &error) {
+
+            });
+        }
+        Q_ASSERT(values.isEmpty());
+    }
+
+};
+
+QTEST_MAIN(MessageQueueTest)
+#include "messagequeuetest.moc"