summaryrefslogtreecommitdiffstats
path: root/common/messagequeue.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'common/messagequeue.cpp')
-rw-r--r--common/messagequeue.cpp24
1 files changed, 12 insertions, 12 deletions
diff --git a/common/messagequeue.cpp b/common/messagequeue.cpp
index 8e3d7d7..1055922 100644
--- a/common/messagequeue.cpp
+++ b/common/messagequeue.cpp
@@ -37,7 +37,7 @@ static KAsync::Job<void> waitForCompletion(QList<KAsync::Future<void> > &futures
37} 37}
38 38
39MessageQueue::MessageQueue(const QString &storageRoot, const QString &name) 39MessageQueue::MessageQueue(const QString &storageRoot, const QString &name)
40 : mStorage(storageRoot, name, Akonadi2::Storage::ReadWrite) 40 : mStorage(storageRoot, name, Sink::Storage::ReadWrite)
41{ 41{
42} 42}
43 43
@@ -56,13 +56,13 @@ void MessageQueue::startTransaction()
56 return; 56 return;
57 } 57 }
58 processRemovals(); 58 processRemovals();
59 mWriteTransaction = std::move(mStorage.createTransaction(Akonadi2::Storage::ReadWrite)); 59 mWriteTransaction = std::move(mStorage.createTransaction(Sink::Storage::ReadWrite));
60} 60}
61 61
62void MessageQueue::commit() 62void MessageQueue::commit()
63{ 63{
64 mWriteTransaction.commit(); 64 mWriteTransaction.commit();
65 mWriteTransaction = Akonadi2::Storage::Transaction(); 65 mWriteTransaction = Sink::Storage::Transaction();
66 processRemovals(); 66 processRemovals();
67 emit messageReady(); 67 emit messageReady();
68} 68}
@@ -74,10 +74,10 @@ void MessageQueue::enqueue(const QByteArray &value)
74 implicitTransaction = true; 74 implicitTransaction = true;
75 startTransaction(); 75 startTransaction();
76 } 76 }
77 const qint64 revision = Akonadi2::Storage::maxRevision(mWriteTransaction) + 1; 77 const qint64 revision = Sink::Storage::maxRevision(mWriteTransaction) + 1;
78 const QByteArray key = QString("%1").arg(revision).toUtf8(); 78 const QByteArray key = QString("%1").arg(revision).toUtf8();
79 mWriteTransaction.openDatabase().write(key, value); 79 mWriteTransaction.openDatabase().write(key, value);
80 Akonadi2::Storage::setMaxRevision(mWriteTransaction, revision); 80 Sink::Storage::setMaxRevision(mWriteTransaction, revision);
81 if (implicitTransaction) { 81 if (implicitTransaction) {
82 commit(); 82 commit();
83 } 83 }
@@ -88,7 +88,7 @@ void MessageQueue::processRemovals()
88 if (mWriteTransaction) { 88 if (mWriteTransaction) {
89 return; 89 return;
90 } 90 }
91 auto transaction = std::move(mStorage.createTransaction(Akonadi2::Storage::ReadWrite)); 91 auto transaction = std::move(mStorage.createTransaction(Sink::Storage::ReadWrite));
92 for (const auto &key : mPendingRemoval) { 92 for (const auto &key : mPendingRemoval) {
93 transaction.openDatabase().remove(key); 93 transaction.openDatabase().remove(key);
94 } 94 }
@@ -117,8 +117,8 @@ KAsync::Job<void> MessageQueue::dequeueBatch(int maxBatchSize, const std::functi
117 return KAsync::start<void>([this, maxBatchSize, resultHandler, resultCount](KAsync::Future<void> &future) { 117 return KAsync::start<void>([this, maxBatchSize, resultHandler, resultCount](KAsync::Future<void> &future) {
118 int count = 0; 118 int count = 0;
119 QList<KAsync::Future<void> > waitCondition; 119 QList<KAsync::Future<void> > waitCondition;
120 mStorage.createTransaction(Akonadi2::Storage::ReadOnly).openDatabase().scan("", [this, resultHandler, resultCount, &count, maxBatchSize, &waitCondition](const QByteArray &key, const QByteArray &value) -> bool { 120 mStorage.createTransaction(Sink::Storage::ReadOnly).openDatabase().scan("", [this, resultHandler, resultCount, &count, maxBatchSize, &waitCondition](const QByteArray &key, const QByteArray &value) -> bool {
121 if (Akonadi2::Storage::isInternalKey(key) || mPendingRemoval.contains(key)) { 121 if (Sink::Storage::isInternalKey(key) || mPendingRemoval.contains(key)) {
122 return true; 122 return true;
123 } 123 }
124 *resultCount += 1; 124 *resultCount += 1;
@@ -133,7 +133,7 @@ KAsync::Job<void> MessageQueue::dequeueBatch(int maxBatchSize, const std::functi
133 } 133 }
134 return false; 134 return false;
135 }, 135 },
136 [](const Akonadi2::Storage::Error &error) { 136 [](const Sink::Storage::Error &error) {
137 ErrorMsg() << "Error while retrieving value" << error.message; 137 ErrorMsg() << "Error while retrieving value" << error.message;
138 // errorHandler(Error(error.store, error.code, error.message)); 138 // errorHandler(Error(error.store, error.code, error.message));
139 }); 139 });
@@ -157,17 +157,17 @@ KAsync::Job<void> MessageQueue::dequeueBatch(int maxBatchSize, const std::functi
157bool MessageQueue::isEmpty() 157bool MessageQueue::isEmpty()
158{ 158{
159 int count = 0; 159 int count = 0;
160 auto t = mStorage.createTransaction(Akonadi2::Storage::ReadOnly); 160 auto t = mStorage.createTransaction(Sink::Storage::ReadOnly);
161 auto db = t.openDatabase(); 161 auto db = t.openDatabase();
162 if (db) { 162 if (db) {
163 db.scan("", [&count, this](const QByteArray &key, const QByteArray &value) -> bool { 163 db.scan("", [&count, this](const QByteArray &key, const QByteArray &value) -> bool {
164 if (!Akonadi2::Storage::isInternalKey(key) && !mPendingRemoval.contains(key)) { 164 if (!Sink::Storage::isInternalKey(key) && !mPendingRemoval.contains(key)) {
165 count++; 165 count++;
166 return false; 166 return false;
167 } 167 }
168 return true; 168 return true;
169 }, 169 },
170 [](const Akonadi2::Storage::Error &error) { 170 [](const Sink::Storage::Error &error) {
171 ErrorMsg() << "Error while checking if empty" << error.message; 171 ErrorMsg() << "Error while checking if empty" << error.message;
172 }); 172 });
173 } 173 }