From a9dc9ed667f06fa1828773d1bb8671ec2731dce5 Mon Sep 17 00:00:00 2001
From: Christian Mollekopf <chrigi_1@fastmail.fm>
Date: Tue, 11 Aug 2015 10:30:10 +0200
Subject: Fixed messagequeue

---
 common/messagequeue.cpp | 21 +++++++++++++++------
 1 file changed, 15 insertions(+), 6 deletions(-)

(limited to 'common/messagequeue.cpp')

diff --git a/common/messagequeue.cpp b/common/messagequeue.cpp
index a92d6be..ecc4d1a 100644
--- a/common/messagequeue.cpp
+++ b/common/messagequeue.cpp
@@ -11,29 +11,38 @@ MessageQueue::MessageQueue(const QString &storageRoot, const QString &name)
 
 void MessageQueue::enqueue(void const *msg, size_t size)
 {
-    auto transaction = mStorage.createTransaction(Akonadi2::Storage::ReadWrite);
-    const qint64 revision = mStorage.maxRevision() + 1;
+    auto transaction = std::move(mStorage.createTransaction(Akonadi2::Storage::ReadWrite));
+    const qint64 revision = Akonadi2::Storage::maxRevision(transaction) + 1;
     const QByteArray key = QString("%1").arg(revision).toUtf8();
     transaction.write(key, QByteArray::fromRawData(static_cast<const char*>(msg), size));
     Akonadi2::Storage::setMaxRevision(transaction, revision);
+    transaction.commit();
     emit messageReady();
 }
 
+void MessageQueue::enqueue(const QByteArray &value)
+{
+    enqueue(value.data(), value.size());
+}
+
 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)
 {
     bool readValue = false;
-    mTransaction = std::move(mStorage.createTransaction(Akonadi2::Storage::ReadWrite));
-    mTransaction.scan("", [this, resultHandler, errorHandler, &readValue](const QByteArray &key, const QByteArray &value) -> bool {
+    auto readTransaction = std::move(mStorage.createTransaction(Akonadi2::Storage::ReadOnly));
+    readTransaction.scan("", [this, resultHandler, errorHandler, &readValue, &readTransaction](const QByteArray &key, const QByteArray &value) -> bool {
         if (Akonadi2::Storage::isInternalKey(key)) {
             return true;
         }
         readValue = true;
         //We need a copy of the key here, otherwise we can't store it in the lambda (the pointers will become invalid)
         const auto keyCopy  = QByteArray(key.constData(), key.size());
-        resultHandler(const_cast<void*>(static_cast<const void*>(value.data())), value.size(), [this, keyCopy, errorHandler](bool success) {
+        //TODO The value copy and the early transaction abort is necessary because we don't support parallel read-transactions yet (in case of a synchronous callback)
+        const auto valueCopy  = QByteArray(value.constData(), value.size());
+        readTransaction.abort();
+        resultHandler(const_cast<void*>(static_cast<const void*>(valueCopy.data())), valueCopy.size(), [this, keyCopy, errorHandler](bool success) {
             if (success) {
-                mTransaction.remove(keyCopy, [errorHandler, keyCopy](const Akonadi2::Storage::Error &error) {
+                mStorage.createTransaction(Akonadi2::Storage::ReadWrite).remove(keyCopy, [errorHandler, keyCopy](const Akonadi2::Storage::Error &error) {
                     ErrorMsg() << "Error while removing value" << error.message << keyCopy;
                     //Don't call the errorhandler in here, we already called the result handler
                 });
-- 
cgit v1.2.3