1 files changed, 47 insertions, 0 deletions

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;
+}
+