summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--common/CMakeLists.txt2
-rw-r--r--common/adaptorfactoryregistry.cpp68
-rw-r--r--common/adaptorfactoryregistry.h64
-rw-r--r--common/changereplay.cpp102
-rw-r--r--common/changereplay.h68
-rw-r--r--common/genericresource.cpp560
-rw-r--r--common/genericresource.h151
-rw-r--r--common/pipeline.cpp17
-rw-r--r--common/pipeline.h3
-rw-r--r--common/resource.cpp2
-rw-r--r--common/resource.h2
-rw-r--r--examples/dummyresource/resourcefactory.cpp163
-rw-r--r--examples/dummyresource/resourcefactory.h6
-rw-r--r--examples/maildirresource/maildirresource.h1
-rw-r--r--tests/dummyresourcetest.cpp2
-rw-r--r--tests/mailquerybenchmark.cpp5
-rw-r--r--tests/pipelinebenchmark.cpp6
-rw-r--r--tests/pipelinetest.cpp14
-rw-r--r--tests/querytest.cpp2
-rw-r--r--tests/testimplementations.h2
20 files changed, 854 insertions, 386 deletions
diff --git a/common/CMakeLists.txt b/common/CMakeLists.txt
index c269a85..79b627a 100644
--- a/common/CMakeLists.txt
+++ b/common/CMakeLists.txt
@@ -66,6 +66,8 @@ set(command_SRCS
66 domain/folder.cpp 66 domain/folder.cpp
67 test.cpp 67 test.cpp
68 query.cpp 68 query.cpp
69 changereplay.cpp
70 adaptorfactoryregistry.cpp
69 ${storage_SRCS}) 71 ${storage_SRCS})
70 72
71add_library(${PROJECT_NAME} SHARED ${command_SRCS}) 73add_library(${PROJECT_NAME} SHARED ${command_SRCS})
diff --git a/common/adaptorfactoryregistry.cpp b/common/adaptorfactoryregistry.cpp
new file mode 100644
index 0000000..323a02d
--- /dev/null
+++ b/common/adaptorfactoryregistry.cpp
@@ -0,0 +1,68 @@
1/*
2 * Copyright (C) 2016 Christian Mollekopf <chrigi_1@fastmail.fm>
3 *
4 * This library is free software; you can redistribute it and/or
5 * modify it under the terms of the GNU Lesser General Public
6 * License as published by the Free Software Foundation; either
7 * version 2.1 of the License, or (at your option) version 3, or any
8 * later version accepted by the membership of KDE e.V. (or its
9 * successor approved by the membership of KDE e.V.), which shall
10 * act as a proxy defined in Section 6 of version 3 of the license.
11 *
12 * This library is distributed in the hope that it will be useful,
13 * but WITHOUT ANY WARRANTY; without even the implied warranty of
14 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
15 * Lesser General Public License for more details.
16 *
17 * You should have received a copy of the GNU Lesser General Public
18 * License along with this library. If not, see <http://www.gnu.org/licenses/>.
19 */
20#include "adaptorfactoryregistry.h"
21
22#include <QByteArray>
23#include <QDebug>
24#include <QMutex>
25#include <functional>
26#include <memory>
27
28#include "domaintypeadaptorfactoryinterface.h"
29#include "applicationdomaintype.h"
30#include "log.h"
31
32using namespace Sink;
33
34AdaptorFactoryRegistry &AdaptorFactoryRegistry::instance()
35{
36 // QMutexLocker locker(&sMutex);
37 static AdaptorFactoryRegistry *instance = 0;
38 if (!instance) {
39 instance = new AdaptorFactoryRegistry;
40 }
41 return *instance;
42}
43
44static QByteArray key(const QByteArray &resource, const QByteArray &type)
45{
46 return resource + type;
47}
48
49AdaptorFactoryRegistry::AdaptorFactoryRegistry()
50{
51
52}
53
54std::shared_ptr<DomainTypeAdaptorFactoryInterface> AdaptorFactoryRegistry::getFactory(const QByteArray &resource, const QByteArray &typeName)
55{
56 const auto ptr = mRegistry.value(key(resource, typeName));
57 //We have to check the pointer before the cast, otherwise a check would return true also for invalid instances.
58 if (!ptr) {
59 return std::shared_ptr<DomainTypeAdaptorFactoryInterface>();
60 }
61 return std::static_pointer_cast<DomainTypeAdaptorFactoryInterface>(ptr);
62}
63
64void AdaptorFactoryRegistry::registerFactory(const QByteArray &resource, const std::shared_ptr<void> &instance, const QByteArray typeName)
65{
66 mRegistry.insert(key(resource, typeName), instance);
67}
68
diff --git a/common/adaptorfactoryregistry.h b/common/adaptorfactoryregistry.h
new file mode 100644
index 0000000..f06120a
--- /dev/null
+++ b/common/adaptorfactoryregistry.h
@@ -0,0 +1,64 @@
1/*
2 * Copyright (C) 2016 Christian Mollekopf <chrigi_1@fastmail.fm>
3 *
4 * This library is free software; you can redistribute it and/or
5 * modify it under the terms of the GNU Lesser General Public
6 * License as published by the Free Software Foundation; either
7 * version 2.1 of the License, or (at your option) version 3, or any
8 * later version accepted by the membership of KDE e.V. (or its
9 * successor approved by the membership of KDE e.V.), which shall
10 * act as a proxy defined in Section 6 of version 3 of the license.
11 *
12 * This library is distributed in the hope that it will be useful,
13 * but WITHOUT ANY WARRANTY; without even the implied warranty of
14 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
15 * Lesser General Public License for more details.
16 *
17 * You should have received a copy of the GNU Lesser General Public
18 * License along with this library. If not, see <http://www.gnu.org/licenses/>.
19 */
20
21#pragma once
22
23#include "sink_export.h"
24#include <QByteArray>
25#include <QDebug>
26#include <QMutex>
27#include <functional>
28#include <memory>
29
30#include "domaintypeadaptorfactoryinterface.h"
31#include "applicationdomaintype.h"
32#include "log.h"
33
34namespace Sink {
35
36/**
37 */
38class SINK_EXPORT AdaptorFactoryRegistry
39{
40public:
41 static AdaptorFactoryRegistry &instance();
42
43 template <class DomainType, class Factory>
44 void registerFactory(const QByteArray &resource)
45 {
46 registerFactory(resource, std::make_shared<Factory>(), ApplicationDomain::getTypeName<DomainType>());
47 }
48
49 template <class DomainType>
50 std::shared_ptr<DomainTypeAdaptorFactoryInterface> getFactory(const QByteArray &resource)
51 {
52 return getFactory(resource, ApplicationDomain::getTypeName<DomainType>());
53 }
54
55 std::shared_ptr<DomainTypeAdaptorFactoryInterface> getFactory(const QByteArray &resource, const QByteArray &typeName);
56
57private:
58 AdaptorFactoryRegistry();
59 void registerFactory(const QByteArray &resource, const std::shared_ptr<void> &instance, const QByteArray typeName);
60
61 QHash<QByteArray, std::shared_ptr<void>> mRegistry;
62 static QMutex sMutex;
63};
64}
diff --git a/common/changereplay.cpp b/common/changereplay.cpp
new file mode 100644
index 0000000..2447b6e
--- /dev/null
+++ b/common/changereplay.cpp
@@ -0,0 +1,102 @@
1/*
2 * Copyright (C) 2016 Christian Mollekopf <mollekopf@kolabsys.com>
3 *
4 * This library is free software; you can redistribute it and/or
5 * modify it under the terms of the GNU Lesser General Public
6 * License as published by the Free Software Foundation; either
7 * version 2.1 of the License, or (at your option) version 3, or any
8 * later version accepted by the membership of KDE e.V. (or its
9 * successor approved by the membership of KDE e.V.), which shall
10 * act as a proxy defined in Section 6 of version 3 of the license.
11 *
12 * This library is distributed in the hope that it will be useful,
13 * but WITHOUT ANY WARRANTY; without even the implied warranty of
14 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
15 * Lesser General Public License for more details.
16 *
17 * You should have received a copy of the GNU Lesser General Public
18 * License along with this library. If not, see <http://www.gnu.org/licenses/>.
19 */
20#include "changereplay.h"
21
22#include "entitybuffer.h"
23#include "log.h"
24#include "definitions.h"
25#include "bufferutils.h"
26
27using namespace Sink;
28
29#undef DEBUG_AREA
30#define DEBUG_AREA "resource.changereplay"
31
32ChangeReplay::ChangeReplay(const QByteArray &resourceName)
33 : mStorage(storageLocation(), resourceName, Storage::ReadOnly), mChangeReplayStore(storageLocation(), resourceName + ".changereplay", Storage::ReadWrite)
34{
35 Trace() << "Created change replay: " << resourceName;
36}
37
38qint64 ChangeReplay::getLastReplayedRevision()
39{
40 qint64 lastReplayedRevision = 0;
41 auto replayStoreTransaction = mChangeReplayStore.createTransaction(Storage::ReadOnly);
42 replayStoreTransaction.openDatabase().scan("lastReplayedRevision",
43 [&lastReplayedRevision](const QByteArray &key, const QByteArray &value) -> bool {
44 lastReplayedRevision = value.toLongLong();
45 return false;
46 },
47 [](const Storage::Error &) {});
48 return lastReplayedRevision;
49}
50
51bool ChangeReplay::allChangesReplayed()
52{
53 const qint64 topRevision = Storage::maxRevision(mStorage.createTransaction(Storage::ReadOnly, [](const Sink::Storage::Error &error) {
54 Warning() << error.message;
55 }));
56 const qint64 lastReplayedRevision = getLastReplayedRevision();
57 Trace() << "All changes replayed " << topRevision << lastReplayedRevision;
58 return (lastReplayedRevision >= topRevision);
59}
60
61void ChangeReplay::revisionChanged()
62{
63 auto mainStoreTransaction = mStorage.createTransaction(Storage::ReadOnly, [](const Sink::Storage::Error &error) {
64 Warning() << error.message;
65 });
66 auto replayStoreTransaction = mChangeReplayStore.createTransaction(Storage::ReadWrite, [](const Sink::Storage::Error &error) {
67 Warning() << error.message;
68 });
69 qint64 lastReplayedRevision = 1;
70 replayStoreTransaction.openDatabase().scan("lastReplayedRevision",
71 [&lastReplayedRevision](const QByteArray &key, const QByteArray &value) -> bool {
72 lastReplayedRevision = value.toLongLong();
73 return false;
74 },
75 [](const Storage::Error &) {});
76 const qint64 topRevision = Storage::maxRevision(mainStoreTransaction);
77
78 Trace() << "Changereplay from " << lastReplayedRevision << " to " << topRevision;
79 if (lastReplayedRevision <= topRevision) {
80 qint64 revision = lastReplayedRevision;
81 for (; revision <= topRevision; revision++) {
82 const auto uid = Storage::getUidFromRevision(mainStoreTransaction, revision);
83 const auto type = Storage::getTypeFromRevision(mainStoreTransaction, revision);
84 const auto key = Storage::assembleKey(uid, revision);
85 Storage::mainDatabase(mainStoreTransaction, type)
86 .scan(key,
87 [&lastReplayedRevision, type, this](const QByteArray &key, const QByteArray &value) -> bool {
88 replay(type, key, value).exec();
89 // TODO make for loop async, and pass to async replay function together with type
90 Trace() << "Replaying " << key;
91 return false;
92 },
93 [key](const Storage::Error &) { ErrorMsg() << "Failed to replay change " << key; });
94 }
95 revision--;
96 replayStoreTransaction.openDatabase().write("lastReplayedRevision", QByteArray::number(revision));
97 replayStoreTransaction.commit();
98 Trace() << "Replayed until " << revision;
99 }
100 emit changesReplayed();
101}
102
diff --git a/common/changereplay.h b/common/changereplay.h
new file mode 100644
index 0000000..a568060
--- /dev/null
+++ b/common/changereplay.h
@@ -0,0 +1,68 @@
1/*
2 * Copyright (C) 2016 Christian Mollekopf <mollekopf@kolabsys.com>
3 *
4 * This library is free software; you can redistribute it and/or
5 * modify it under the terms of the GNU Lesser General Public
6 * License as published by the Free Software Foundation; either
7 * version 2.1 of the License, or (at your option) version 3, or any
8 * later version accepted by the membership of KDE e.V. (or its
9 * successor approved by the membership of KDE e.V.), which shall
10 * act as a proxy defined in Section 6 of version 3 of the license.
11 *
12 * This library is distributed in the hope that it will be useful,
13 * but WITHOUT ANY WARRANTY; without even the implied warranty of
14 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
15 * Lesser General Public License for more details.
16 *
17 * You should have received a copy of the GNU Lesser General Public
18 * License along with this library. If not, see <http://www.gnu.org/licenses/>.
19 */
20#pragma once
21
22#include "sink_export.h"
23#include <QObject>
24#include <Async/Async>
25
26#include "storage.h"
27
28namespace Sink {
29
30/**
31 * Replays changes from the storage one by one.
32 *
33 * Uses a local database to:
34 * * Remember what changes have been replayed already.
35 * * store a mapping of remote to local buffers
36 */
37class SINK_EXPORT ChangeReplay : public QObject
38{
39 Q_OBJECT
40public:
41 ChangeReplay(const QByteArray &resourceName);
42
43 qint64 getLastReplayedRevision();
44 bool allChangesReplayed();
45
46signals:
47 void changesReplayed();
48
49public slots:
50 void revisionChanged();
51
52protected:
53 virtual KAsync::Job<void> replay(const QByteArray &type, const QByteArray &key, const QByteArray &value) = 0;
54 Sink::Storage mStorage;
55
56private:
57 Sink::Storage mChangeReplayStore;
58};
59
60class NullChangeReplay : public ChangeReplay
61{
62 public:
63 NullChangeReplay() : ChangeReplay("null") {}
64 KAsync::Job<void> replay(const QByteArray &type, const QByteArray &key, const QByteArray &value) Q_DECL_OVERRIDE { return KAsync::null<void>(); }
65};
66
67}
68
diff --git a/common/genericresource.cpp b/common/genericresource.cpp
index eae6ead..cb2ef21 100644
--- a/common/genericresource.cpp
+++ b/common/genericresource.cpp
@@ -1,3 +1,22 @@
1/*
2 * Copyright (C) 2016 Christian Mollekopf <mollekopf@kolabsys.com>
3 *
4 * This library is free software; you can redistribute it and/or
5 * modify it under the terms of the GNU Lesser General Public
6 * License as published by the Free Software Foundation; either
7 * version 2.1 of the License, or (at your option) version 3, or any
8 * later version accepted by the membership of KDE e.V. (or its
9 * successor approved by the membership of KDE e.V.), which shall
10 * act as a proxy defined in Section 6 of version 3 of the license.
11 *
12 * This library is distributed in the hope that it will be useful,
13 * but WITHOUT ANY WARRANTY; without even the implied warranty of
14 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
15 * Lesser General Public License for more details.
16 *
17 * You should have received a copy of the GNU Lesser General Public
18 * License along with this library. If not, see <http://www.gnu.org/licenses/>.
19 */
1#include "genericresource.h" 20#include "genericresource.h"
2 21
3#include "entitybuffer.h" 22#include "entitybuffer.h"
@@ -14,6 +33,7 @@
14#include "log.h" 33#include "log.h"
15#include "definitions.h" 34#include "definitions.h"
16#include "bufferutils.h" 35#include "bufferutils.h"
36#include "adaptorfactoryregistry.h"
17 37
18#include <QUuid> 38#include <QUuid>
19#include <QDataStream> 39#include <QDataStream>
@@ -30,96 +50,6 @@ static int sCommitInterval = 10;
30using namespace Sink; 50using namespace Sink;
31 51
32#undef DEBUG_AREA 52#undef DEBUG_AREA
33#define DEBUG_AREA "resource.changereplay"
34
35/**
36 * Replays changes from the storage one by one.
37 *
38 * Uses a local database to:
39 * * Remember what changes have been replayed already.
40 * * store a mapping of remote to local buffers
41 */
42class ChangeReplay : public QObject
43{
44 Q_OBJECT
45public:
46 typedef std::function<KAsync::Job<void>(const QByteArray &type, const QByteArray &key, const QByteArray &value)> ReplayFunction;
47
48 ChangeReplay(const QString &resourceName, const ReplayFunction &replayFunction)
49 : mStorage(storageLocation(), resourceName, Storage::ReadOnly), mChangeReplayStore(storageLocation(), resourceName + ".changereplay", Storage::ReadWrite), mReplayFunction(replayFunction)
50 {
51 }
52
53 qint64 getLastReplayedRevision()
54 {
55 qint64 lastReplayedRevision = 0;
56 auto replayStoreTransaction = mChangeReplayStore.createTransaction(Storage::ReadOnly);
57 replayStoreTransaction.openDatabase().scan("lastReplayedRevision",
58 [&lastReplayedRevision](const QByteArray &key, const QByteArray &value) -> bool {
59 lastReplayedRevision = value.toLongLong();
60 return false;
61 },
62 [](const Storage::Error &) {});
63 return lastReplayedRevision;
64 }
65
66 bool allChangesReplayed()
67 {
68 const qint64 topRevision = Storage::maxRevision(mStorage.createTransaction(Storage::ReadOnly));
69 const qint64 lastReplayedRevision = getLastReplayedRevision();
70 Trace() << "All changes replayed " << topRevision << lastReplayedRevision;
71 return (lastReplayedRevision >= topRevision);
72 }
73
74signals:
75 void changesReplayed();
76
77public slots:
78 void revisionChanged()
79 {
80 auto mainStoreTransaction = mStorage.createTransaction(Storage::ReadOnly);
81 auto replayStoreTransaction = mChangeReplayStore.createTransaction(Storage::ReadWrite);
82 qint64 lastReplayedRevision = 1;
83 replayStoreTransaction.openDatabase().scan("lastReplayedRevision",
84 [&lastReplayedRevision](const QByteArray &key, const QByteArray &value) -> bool {
85 lastReplayedRevision = value.toLongLong();
86 return false;
87 },
88 [](const Storage::Error &) {});
89 const qint64 topRevision = Storage::maxRevision(mainStoreTransaction);
90
91 Trace() << "Changereplay from " << lastReplayedRevision << " to " << topRevision;
92 if (lastReplayedRevision <= topRevision) {
93 qint64 revision = lastReplayedRevision;
94 for (; revision <= topRevision; revision++) {
95 const auto uid = Storage::getUidFromRevision(mainStoreTransaction, revision);
96 const auto type = Storage::getTypeFromRevision(mainStoreTransaction, revision);
97 const auto key = Storage::assembleKey(uid, revision);
98 Storage::mainDatabase(mainStoreTransaction, type)
99 .scan(key,
100 [&lastReplayedRevision, type, this](const QByteArray &key, const QByteArray &value) -> bool {
101 mReplayFunction(type, key, value).exec();
102 // TODO make for loop async, and pass to async replay function together with type
103 Trace() << "Replaying " << key;
104 return false;
105 },
106 [key](const Storage::Error &) { ErrorMsg() << "Failed to replay change " << key; });
107 }
108 revision--;
109 replayStoreTransaction.openDatabase().write("lastReplayedRevision", QByteArray::number(revision));
110 replayStoreTransaction.commit();
111 Trace() << "Replayed until " << revision;
112 }
113 emit changesReplayed();
114 }
115
116private:
117 Sink::Storage mStorage;
118 Sink::Storage mChangeReplayStore;
119 ReplayFunction mReplayFunction;
120};
121
122#undef DEBUG_AREA
123#define DEBUG_AREA "resource.commandprocessor" 53#define DEBUG_AREA "resource.commandprocessor"
124 54
125/** 55/**
@@ -133,10 +63,9 @@ class CommandProcessor : public QObject
133public: 63public:
134 CommandProcessor(Sink::Pipeline *pipeline, QList<MessageQueue *> commandQueues) : QObject(), mPipeline(pipeline), mCommandQueues(commandQueues), mProcessingLock(false) 64 CommandProcessor(Sink::Pipeline *pipeline, QList<MessageQueue *> commandQueues) : QObject(), mPipeline(pipeline), mCommandQueues(commandQueues), mProcessingLock(false)
135 { 65 {
136 mPipeline->startTransaction(); 66 mLowerBoundRevision = Storage::maxRevision(mPipeline->storage().createTransaction(Storage::ReadOnly, [](const Sink::Storage::Error &error) {
137 // FIXME Should be initialized to the current value of the change replay queue 67 Warning() << error.message;
138 mLowerBoundRevision = Storage::maxRevision(mPipeline->transaction()); 68 }));
139 mPipeline->commit();
140 69
141 for (auto queue : mCommandQueues) { 70 for (auto queue : mCommandQueues) {
142 const bool ret = connect(queue, &MessageQueue::messageReady, this, &CommandProcessor::process); 71 const bool ret = connect(queue, &MessageQueue::messageReady, this, &CommandProcessor::process);
@@ -303,15 +232,22 @@ private:
303#undef DEBUG_AREA 232#undef DEBUG_AREA
304#define DEBUG_AREA "resource" 233#define DEBUG_AREA "resource"
305 234
306GenericResource::GenericResource(const QByteArray &resourceInstanceIdentifier, const QSharedPointer<Pipeline> &pipeline) 235GenericResource::GenericResource(const QByteArray &resourceType, const QByteArray &resourceInstanceIdentifier, const QSharedPointer<Pipeline> &pipeline, const QSharedPointer<ChangeReplay> &changeReplay, const QSharedPointer<Synchronizer> &synchronizer)
307 : Sink::Resource(), 236 : Sink::Resource(),
308 mUserQueue(Sink::storageLocation(), resourceInstanceIdentifier + ".userqueue"), 237 mUserQueue(Sink::storageLocation(), resourceInstanceIdentifier + ".userqueue"),
309 mSynchronizerQueue(Sink::storageLocation(), resourceInstanceIdentifier + ".synchronizerqueue"), 238 mSynchronizerQueue(Sink::storageLocation(), resourceInstanceIdentifier + ".synchronizerqueue"),
239 mResourceType(resourceType),
310 mResourceInstanceIdentifier(resourceInstanceIdentifier), 240 mResourceInstanceIdentifier(resourceInstanceIdentifier),
311 mPipeline(pipeline ? pipeline : QSharedPointer<Sink::Pipeline>::create(resourceInstanceIdentifier)), 241 mPipeline(pipeline ? pipeline : QSharedPointer<Sink::Pipeline>::create(resourceInstanceIdentifier)),
242 mChangeReplay(changeReplay),
243 mSynchronizer(synchronizer),
312 mError(0), 244 mError(0),
313 mClientLowerBoundRevision(std::numeric_limits<qint64>::max()) 245 mClientLowerBoundRevision(std::numeric_limits<qint64>::max())
314{ 246{
247 mPipeline->setResourceType(mResourceType);
248 mSynchronizer->setup([this](int commandId, const QByteArray &data) {
249 enqueueCommand(mSynchronizerQueue, commandId, data);
250 });
315 mProcessor = new CommandProcessor(mPipeline.data(), QList<MessageQueue *>() << &mUserQueue << &mSynchronizerQueue); 251 mProcessor = new CommandProcessor(mPipeline.data(), QList<MessageQueue *>() << &mUserQueue << &mSynchronizerQueue);
316 mProcessor->setInspectionCommand([this](void const *command, size_t size) { 252 mProcessor->setInspectionCommand([this](void const *command, size_t size) {
317 flatbuffers::Verifier verifier((const uint8_t *)command, size); 253 flatbuffers::Verifier verifier((const uint8_t *)command, size);
@@ -353,14 +289,9 @@ GenericResource::GenericResource(const QByteArray &resourceInstanceIdentifier, c
353 }); 289 });
354 QObject::connect(mProcessor, &CommandProcessor::error, [this](int errorCode, const QString &msg) { onProcessorError(errorCode, msg); }); 290 QObject::connect(mProcessor, &CommandProcessor::error, [this](int errorCode, const QString &msg) { onProcessorError(errorCode, msg); });
355 QObject::connect(mPipeline.data(), &Pipeline::revisionUpdated, this, &Resource::revisionUpdated); 291 QObject::connect(mPipeline.data(), &Pipeline::revisionUpdated, this, &Resource::revisionUpdated);
356 mSourceChangeReplay = new ChangeReplay(resourceInstanceIdentifier, [this](const QByteArray &type, const QByteArray &key, const QByteArray &value) {
357 // This results in a deadlock when a sync is in progress and we try to create a second writing transaction (which is why we turn changereplay off during the sync)
358 auto synchronizationStore = QSharedPointer<Sink::Storage>::create(Sink::storageLocation(), mResourceInstanceIdentifier + ".synchronization", Sink::Storage::ReadWrite);
359 return this->replay(*synchronizationStore, type, key, value).then<void>([synchronizationStore]() {});
360 });
361 enableChangeReplay(true); 292 enableChangeReplay(true);
362 mClientLowerBoundRevision = mPipeline->cleanedUpRevision(); 293 mClientLowerBoundRevision = mPipeline->cleanedUpRevision();
363 mProcessor->setOldestUsedRevision(mSourceChangeReplay->getLastReplayedRevision()); 294 mProcessor->setOldestUsedRevision(mChangeReplay->getLastReplayedRevision());
364 295
365 mCommitQueueTimer.setInterval(sCommitInterval); 296 mCommitQueueTimer.setInterval(sCommitInterval);
366 mCommitQueueTimer.setSingleShot(true); 297 mCommitQueueTimer.setSingleShot(true);
@@ -370,7 +301,6 @@ GenericResource::GenericResource(const QByteArray &resourceInstanceIdentifier, c
370GenericResource::~GenericResource() 301GenericResource::~GenericResource()
371{ 302{
372 delete mProcessor; 303 delete mProcessor;
373 delete mSourceChangeReplay;
374} 304}
375 305
376KAsync::Job<void> GenericResource::inspect( 306KAsync::Job<void> GenericResource::inspect(
@@ -383,86 +313,20 @@ KAsync::Job<void> GenericResource::inspect(
383void GenericResource::enableChangeReplay(bool enable) 313void GenericResource::enableChangeReplay(bool enable)
384{ 314{
385 if (enable) { 315 if (enable) {
386 QObject::connect(mPipeline.data(), &Pipeline::revisionUpdated, mSourceChangeReplay, &ChangeReplay::revisionChanged, Qt::QueuedConnection); 316 QObject::connect(mPipeline.data(), &Pipeline::revisionUpdated, mChangeReplay.data(), &ChangeReplay::revisionChanged, Qt::QueuedConnection);
387 QObject::connect(mSourceChangeReplay, &ChangeReplay::changesReplayed, this, &GenericResource::updateLowerBoundRevision); 317 QObject::connect(mChangeReplay.data(), &ChangeReplay::changesReplayed, this, &GenericResource::updateLowerBoundRevision);
388 mSourceChangeReplay->revisionChanged(); 318 mChangeReplay->revisionChanged();
389 } else { 319 } else {
390 QObject::disconnect(mPipeline.data(), &Pipeline::revisionUpdated, mSourceChangeReplay, &ChangeReplay::revisionChanged); 320 QObject::disconnect(mPipeline.data(), &Pipeline::revisionUpdated, mChangeReplay.data(), &ChangeReplay::revisionChanged);
391 QObject::disconnect(mSourceChangeReplay, &ChangeReplay::changesReplayed, this, &GenericResource::updateLowerBoundRevision); 321 QObject::disconnect(mChangeReplay.data(), &ChangeReplay::changesReplayed, this, &GenericResource::updateLowerBoundRevision);
392 } 322 }
393} 323}
394 324
395void GenericResource::addType(const QByteArray &type, DomainTypeAdaptorFactoryInterface::Ptr factory, const QVector<Sink::Preprocessor *> &preprocessors) 325void GenericResource::addType(const QByteArray &type, DomainTypeAdaptorFactoryInterface::Ptr factory, const QVector<Sink::Preprocessor *> &preprocessors)
396{ 326{
397 mPipeline->setPreprocessors(type, preprocessors); 327 mPipeline->setPreprocessors(type, preprocessors);
398 mPipeline->setAdaptorFactory(type, factory);
399 mAdaptorFactories.insert(type, factory);
400}
401
402KAsync::Job<void> GenericResource::replay(Sink::Storage &synchronizationStore, const QByteArray &type, const QByteArray &key, const QByteArray &value)
403{
404 Sink::EntityBuffer buffer(value);
405 const Sink::Entity &entity = buffer.entity();
406 const auto metadataBuffer = Sink::EntityBuffer::readBuffer<Sink::Metadata>(entity.metadata());
407 Q_ASSERT(metadataBuffer);
408 if (!metadataBuffer->replayToSource()) {
409 Trace() << "Change is coming from the source";
410 return KAsync::null<void>();
411 }
412 const qint64 revision = metadataBuffer ? metadataBuffer->revision() : -1;
413 const auto operation = metadataBuffer ? metadataBuffer->operation() : Sink::Operation_Creation;
414 const auto uid = Sink::Storage::uidFromKey(key);
415 QByteArray oldRemoteId;
416
417 if (operation != Sink::Operation_Creation) {
418 auto synchronizationTransaction = synchronizationStore.createTransaction(Sink::Storage::ReadOnly);
419 oldRemoteId = resolveLocalId(type, uid, synchronizationTransaction);
420 }
421 Trace() << "Replaying " << key << type;
422
423 KAsync::Job<QByteArray> job = KAsync::null<QByteArray>();
424 if (type == ENTITY_TYPE_FOLDER) {
425 const Sink::ApplicationDomain::Folder folder(mResourceInstanceIdentifier, uid, revision, mAdaptorFactories.value(type)->createAdaptor(entity));
426 job = replay(folder, operation, oldRemoteId);
427 } else if (type == ENTITY_TYPE_MAIL) {
428 const Sink::ApplicationDomain::Mail mail(mResourceInstanceIdentifier, uid, revision, mAdaptorFactories.value(type)->createAdaptor(entity));
429 job = replay(mail, operation, oldRemoteId);
430 }
431
432 return job.then<void, QByteArray>([=, &synchronizationStore](const QByteArray &remoteId) {
433 auto synchronizationTransaction = synchronizationStore.createTransaction(Sink::Storage::ReadWrite);
434 Trace() << "Replayed change with remote id: " << remoteId;
435 if (operation == Sink::Operation_Creation) {
436 if (remoteId.isEmpty()) {
437 Warning() << "Returned an empty remoteId from the creation";
438 } else {
439 recordRemoteId(type, uid, remoteId, synchronizationTransaction);
440 }
441 } else if (operation == Sink::Operation_Modification) {
442 if (remoteId.isEmpty()) {
443 Warning() << "Returned an empty remoteId from the creation";
444 } else {
445 updateRemoteId(type, uid, remoteId, synchronizationTransaction);
446 }
447 } else if (operation == Sink::Operation_Removal) {
448 removeRemoteId(type, uid, remoteId, synchronizationTransaction);
449 } else {
450 Warning() << "Unkown operation" << operation;
451 }
452 }, [](int errorCode, const QString &errorMessage) {
453 Warning() << "Failed to replay change: " << errorMessage;
454 });
455} 328}
456 329
457KAsync::Job<QByteArray> GenericResource::replay(const ApplicationDomain::Mail &, Sink::Operation, const QByteArray &)
458{
459 return KAsync::null<QByteArray>();
460}
461
462KAsync::Job<QByteArray> GenericResource::replay(const ApplicationDomain::Folder &, Sink::Operation, const QByteArray &)
463{
464 return KAsync::null<QByteArray>();
465}
466 330
467void GenericResource::removeDataFromDisk() 331void GenericResource::removeDataFromDisk()
468{ 332{
@@ -528,10 +392,8 @@ KAsync::Job<void> GenericResource::synchronizeWithSource()
528 Log() << " Synchronizing"; 392 Log() << " Synchronizing";
529 // Changereplay would deadlock otherwise when trying to open the synchronization store 393 // Changereplay would deadlock otherwise when trying to open the synchronization store
530 enableChangeReplay(false); 394 enableChangeReplay(false);
531 auto mainStore = QSharedPointer<Sink::Storage>::create(Sink::storageLocation(), mResourceInstanceIdentifier, Sink::Storage::ReadOnly); 395 mSynchronizer->synchronize()
532 auto syncStore = QSharedPointer<Sink::Storage>::create(Sink::storageLocation(), mResourceInstanceIdentifier + ".synchronization", Sink::Storage::ReadWrite); 396 .then<void>([this, &future]() {
533 synchronizeWithSource(*mainStore, *syncStore)
534 .then<void>([this, mainStore, syncStore, &future]() {
535 Log() << "Done Synchronizing"; 397 Log() << "Done Synchronizing";
536 enableChangeReplay(true); 398 enableChangeReplay(true);
537 future.setFinished(); 399 future.setFinished();
@@ -576,11 +438,11 @@ KAsync::Job<void> GenericResource::processAllMessages()
576 .then<void>([this](KAsync::Future<void> &f) { waitForDrained(f, mSynchronizerQueue); }) 438 .then<void>([this](KAsync::Future<void> &f) { waitForDrained(f, mSynchronizerQueue); })
577 .then<void>([this](KAsync::Future<void> &f) { waitForDrained(f, mUserQueue); }) 439 .then<void>([this](KAsync::Future<void> &f) { waitForDrained(f, mUserQueue); })
578 .then<void>([this](KAsync::Future<void> &f) { 440 .then<void>([this](KAsync::Future<void> &f) {
579 if (mSourceChangeReplay->allChangesReplayed()) { 441 if (mChangeReplay->allChangesReplayed()) {
580 f.setFinished(); 442 f.setFinished();
581 } else { 443 } else {
582 auto context = new QObject; 444 auto context = new QObject;
583 QObject::connect(mSourceChangeReplay, &ChangeReplay::changesReplayed, context, [&f, context]() { 445 QObject::connect(mChangeReplay.data(), &ChangeReplay::changesReplayed, context, [&f, context]() {
584 delete context; 446 delete context;
585 f.setFinished(); 447 f.setFinished();
586 }); 448 });
@@ -590,7 +452,7 @@ KAsync::Job<void> GenericResource::processAllMessages()
590 452
591void GenericResource::updateLowerBoundRevision() 453void GenericResource::updateLowerBoundRevision()
592{ 454{
593 mProcessor->setOldestUsedRevision(qMin(mClientLowerBoundRevision, mSourceChangeReplay->getLastReplayedRevision())); 455 mProcessor->setOldestUsedRevision(qMin(mClientLowerBoundRevision, mChangeReplay->getLastReplayedRevision()));
594} 456}
595 457
596void GenericResource::setLowerBoundRevision(qint64 revision) 458void GenericResource::setLowerBoundRevision(qint64 revision)
@@ -599,7 +461,139 @@ void GenericResource::setLowerBoundRevision(qint64 revision)
599 updateLowerBoundRevision(); 461 updateLowerBoundRevision();
600} 462}
601 463
602void GenericResource::createEntity(const QByteArray &sinkId, const QByteArray &bufferType, const Sink::ApplicationDomain::ApplicationDomainType &domainObject, 464
465
466
467EntityStore::EntityStore(const QByteArray &resourceType, const QByteArray &resourceInstanceIdentifier, Sink::Storage::Transaction &transaction)
468 : mResourceType(resourceType), mResourceInstanceIdentifier(resourceInstanceIdentifier),
469 mTransaction(transaction)
470{
471
472}
473
474template<typename T>
475T EntityStore::read(const QByteArray &identifier) const
476{
477 auto typeName = ApplicationDomain::getTypeName<T>();
478 auto mainDatabase = Storage::mainDatabase(mTransaction, typeName);
479 auto bufferAdaptor = getLatest(mainDatabase, identifier, *Sink::AdaptorFactoryRegistry::instance().getFactory<T>(mResourceType));
480 Q_ASSERT(bufferAdaptor);
481 return T(mResourceInstanceIdentifier, identifier, 0, bufferAdaptor);
482}
483
484QSharedPointer<Sink::ApplicationDomain::BufferAdaptor> EntityStore::getLatest(const Sink::Storage::NamedDatabase &db, const QByteArray &uid, DomainTypeAdaptorFactoryInterface &adaptorFactory)
485{
486 QSharedPointer<Sink::ApplicationDomain::BufferAdaptor> current;
487 db.findLatest(uid,
488 [&current, &adaptorFactory](const QByteArray &key, const QByteArray &data) -> bool {
489 Sink::EntityBuffer buffer(const_cast<const char *>(data.data()), data.size());
490 if (!buffer.isValid()) {
491 Warning() << "Read invalid buffer from disk";
492 } else {
493 Trace() << "Found value " << key;
494 current = adaptorFactory.createAdaptor(buffer.entity());
495 }
496 return false;
497 },
498 [](const Sink::Storage::Error &error) { Warning() << "Failed to read current value from storage: " << error.message; });
499 return current;
500}
501
502
503
504SyncStore::SyncStore(Sink::Storage::Transaction &transaction)
505 : mTransaction(transaction)
506{
507
508}
509
510void SyncStore::recordRemoteId(const QByteArray &bufferType, const QByteArray &localId, const QByteArray &remoteId)
511{
512 Index("rid.mapping." + bufferType, mTransaction).add(remoteId, localId);
513 Index("localid.mapping." + bufferType, mTransaction).add(localId, remoteId);
514}
515
516void SyncStore::removeRemoteId(const QByteArray &bufferType, const QByteArray &localId, const QByteArray &remoteId)
517{
518 Index("rid.mapping." + bufferType, mTransaction).remove(remoteId, localId);
519 Index("localid.mapping." + bufferType, mTransaction).remove(localId, remoteId);
520}
521
522void SyncStore::updateRemoteId(const QByteArray &bufferType, const QByteArray &localId, const QByteArray &remoteId)
523{
524 const auto oldRemoteId = Index("localid.mapping." + bufferType, mTransaction).lookup(localId);
525 removeRemoteId(bufferType, localId, oldRemoteId);
526 recordRemoteId(bufferType, localId, remoteId);
527}
528
529QByteArray SyncStore::resolveRemoteId(const QByteArray &bufferType, const QByteArray &remoteId)
530{
531 // Lookup local id for remote id, or insert a new pair otherwise
532 Index index("rid.mapping." + bufferType, mTransaction);
533 QByteArray sinkId = index.lookup(remoteId);
534 if (sinkId.isEmpty()) {
535 sinkId = QUuid::createUuid().toString().toUtf8();
536 index.add(remoteId, sinkId);
537 Index("localid.mapping." + bufferType, mTransaction).add(sinkId, remoteId);
538 }
539 return sinkId;
540}
541
542QByteArray SyncStore::resolveLocalId(const QByteArray &bufferType, const QByteArray &localId)
543{
544 QByteArray remoteId = Index("localid.mapping." + bufferType, mTransaction).lookup(localId);
545 if (remoteId.isEmpty()) {
546 Warning() << "Couldn't find the remote id for " << localId;
547 return QByteArray();
548 }
549 return remoteId;
550}
551
552
553
554
555
556
557
558
559Synchronizer::Synchronizer(const QByteArray &resourceType, const QByteArray &resourceInstanceIdentifier)
560 : mStorage(Sink::storageLocation(), resourceInstanceIdentifier, Sink::Storage::ReadOnly),
561 mSyncStorage(Sink::storageLocation(), resourceInstanceIdentifier + ".synchronization", Sink::Storage::ReadWrite),
562 mResourceType(resourceType),
563 mResourceInstanceIdentifier(resourceInstanceIdentifier)
564{
565 Trace() << "Starting synchronizer: " << resourceType << resourceInstanceIdentifier;
566
567}
568
569void Synchronizer::setup(const std::function<void(int commandId, const QByteArray &data)> &enqueueCommandCallback)
570{
571 mEnqueue = enqueueCommandCallback;
572}
573
574void Synchronizer::enqueueCommand(int commandId, const QByteArray &data)
575{
576 Q_ASSERT(mEnqueue);
577 mEnqueue(commandId, data);
578}
579
580EntityStore &Synchronizer::store()
581{
582 if (!mEntityStore) {
583 mEntityStore = QSharedPointer<EntityStore>::create(mResourceType, mResourceInstanceIdentifier, mTransaction);
584 }
585 return *mEntityStore;
586}
587
588SyncStore &Synchronizer::syncStore()
589{
590 if (!mSyncStore) {
591 mSyncStore = QSharedPointer<SyncStore>::create(mSyncTransaction);
592 }
593 return *mSyncStore;
594}
595
596void Synchronizer::createEntity(const QByteArray &sinkId, const QByteArray &bufferType, const Sink::ApplicationDomain::ApplicationDomainType &domainObject,
603 DomainTypeAdaptorFactoryInterface &adaptorFactory, std::function<void(const QByteArray &)> callback) 597 DomainTypeAdaptorFactoryInterface &adaptorFactory, std::function<void(const QByteArray &)> callback)
604{ 598{
605 // These changes are coming from the source 599 // These changes are coming from the source
@@ -616,7 +610,7 @@ void GenericResource::createEntity(const QByteArray &sinkId, const QByteArray &b
616 callback(BufferUtils::extractBuffer(fbb)); 610 callback(BufferUtils::extractBuffer(fbb));
617} 611}
618 612
619void GenericResource::modifyEntity(const QByteArray &sinkId, qint64 revision, const QByteArray &bufferType, const Sink::ApplicationDomain::ApplicationDomainType &domainObject, 613void Synchronizer::modifyEntity(const QByteArray &sinkId, qint64 revision, const QByteArray &bufferType, const Sink::ApplicationDomain::ApplicationDomainType &domainObject,
620 DomainTypeAdaptorFactoryInterface &adaptorFactory, std::function<void(const QByteArray &)> callback) 614 DomainTypeAdaptorFactoryInterface &adaptorFactory, std::function<void(const QByteArray &)> callback)
621{ 615{
622 // These changes are coming from the source 616 // These changes are coming from the source
@@ -634,7 +628,7 @@ void GenericResource::modifyEntity(const QByteArray &sinkId, qint64 revision, co
634 callback(BufferUtils::extractBuffer(fbb)); 628 callback(BufferUtils::extractBuffer(fbb));
635} 629}
636 630
637void GenericResource::deleteEntity(const QByteArray &sinkId, qint64 revision, const QByteArray &bufferType, std::function<void(const QByteArray &)> callback) 631void Synchronizer::deleteEntity(const QByteArray &sinkId, qint64 revision, const QByteArray &bufferType, std::function<void(const QByteArray &)> callback)
638{ 632{
639 // These changes are coming from the source 633 // These changes are coming from the source
640 const auto replayToSource = false; 634 const auto replayToSource = false;
@@ -647,96 +641,36 @@ void GenericResource::deleteEntity(const QByteArray &sinkId, qint64 revision, co
647 callback(BufferUtils::extractBuffer(fbb)); 641 callback(BufferUtils::extractBuffer(fbb));
648} 642}
649 643
650void GenericResource::recordRemoteId(const QByteArray &bufferType, const QByteArray &localId, const QByteArray &remoteId, Sink::Storage::Transaction &transaction) 644void Synchronizer::scanForRemovals(const QByteArray &bufferType, const std::function<void(const std::function<void(const QByteArray &key)> &callback)> &entryGenerator, std::function<bool(const QByteArray &remoteId)> exists)
651{
652 Index("rid.mapping." + bufferType, transaction).add(remoteId, localId);
653 ;
654 Index("localid.mapping." + bufferType, transaction).add(localId, remoteId);
655}
656
657void GenericResource::removeRemoteId(const QByteArray &bufferType, const QByteArray &localId, const QByteArray &remoteId, Sink::Storage::Transaction &transaction)
658{
659 Index("rid.mapping." + bufferType, transaction).remove(remoteId, localId);
660 Index("localid.mapping." + bufferType, transaction).remove(localId, remoteId);
661}
662
663void GenericResource::updateRemoteId(const QByteArray &bufferType, const QByteArray &localId, const QByteArray &remoteId, Sink::Storage::Transaction &transaction)
664{
665 const auto oldRemoteId = Index("localid.mapping." + bufferType, transaction).lookup(localId);
666 removeRemoteId(bufferType, localId, oldRemoteId, transaction);
667 recordRemoteId(bufferType, localId, remoteId, transaction);
668}
669
670QByteArray GenericResource::resolveRemoteId(const QByteArray &bufferType, const QByteArray &remoteId, Sink::Storage::Transaction &transaction)
671{
672 // Lookup local id for remote id, or insert a new pair otherwise
673 Index index("rid.mapping." + bufferType, transaction);
674 QByteArray sinkId = index.lookup(remoteId);
675 if (sinkId.isEmpty()) {
676 sinkId = QUuid::createUuid().toString().toUtf8();
677 index.add(remoteId, sinkId);
678 Index("localid.mapping." + bufferType, transaction).add(sinkId, remoteId);
679 }
680 return sinkId;
681}
682
683QByteArray GenericResource::resolveLocalId(const QByteArray &bufferType, const QByteArray &localId, Sink::Storage::Transaction &transaction)
684{
685 QByteArray remoteId = Index("localid.mapping." + bufferType, transaction).lookup(localId);
686 if (remoteId.isEmpty()) {
687 Warning() << "Couldn't find the remote id for " << localId;
688 return QByteArray();
689 }
690 return remoteId;
691}
692
693void GenericResource::scanForRemovals(Sink::Storage::Transaction &transaction, Sink::Storage::Transaction &synchronizationTransaction, const QByteArray &bufferType,
694 const std::function<void(const std::function<void(const QByteArray &key)> &callback)> &entryGenerator, std::function<bool(const QByteArray &remoteId)> exists)
695{ 645{
696 entryGenerator([this, &transaction, bufferType, &synchronizationTransaction, &exists](const QByteArray &key) { 646 entryGenerator([this, bufferType, &exists](const QByteArray &key) {
697 auto sinkId = Sink::Storage::uidFromKey(key); 647 auto sinkId = Sink::Storage::uidFromKey(key);
698 Trace() << "Checking for removal " << key; 648 Trace() << "Checking for removal " << key;
699 const auto remoteId = resolveLocalId(bufferType, sinkId, synchronizationTransaction); 649 const auto remoteId = syncStore().resolveLocalId(bufferType, sinkId);
700 // If we have no remoteId, the entity hasn't been replayed to the source yet 650 // If we have no remoteId, the entity hasn't been replayed to the source yet
701 if (!remoteId.isEmpty()) { 651 if (!remoteId.isEmpty()) {
702 if (!exists(remoteId)) { 652 if (!exists(remoteId)) {
703 Trace() << "Found a removed entity: " << sinkId; 653 Trace() << "Found a removed entity: " << sinkId;
704 deleteEntity(sinkId, Sink::Storage::maxRevision(transaction), bufferType, 654 deleteEntity(sinkId, Sink::Storage::maxRevision(mTransaction), bufferType,
705 [this](const QByteArray &buffer) { enqueueCommand(mSynchronizerQueue, Sink::Commands::DeleteEntityCommand, buffer); }); 655 [this](const QByteArray &buffer) { enqueueCommand(Sink::Commands::DeleteEntityCommand, buffer); });
706 } 656 }
707 } 657 }
708 }); 658 });
709} 659}
710 660
711QSharedPointer<Sink::ApplicationDomain::BufferAdaptor> GenericResource::getLatest(const Sink::Storage::NamedDatabase &db, const QByteArray &uid, DomainTypeAdaptorFactoryInterface &adaptorFactory) 661void Synchronizer::createOrModify(const QByteArray &bufferType, const QByteArray &remoteId, const Sink::ApplicationDomain::ApplicationDomainType &entity)
712{
713 QSharedPointer<Sink::ApplicationDomain::BufferAdaptor> current;
714 db.findLatest(uid,
715 [&current, &adaptorFactory](const QByteArray &key, const QByteArray &data) -> bool {
716 Sink::EntityBuffer buffer(const_cast<const char *>(data.data()), data.size());
717 if (!buffer.isValid()) {
718 Warning() << "Read invalid buffer from disk";
719 } else {
720 current = adaptorFactory.createAdaptor(buffer.entity());
721 }
722 return false;
723 },
724 [](const Sink::Storage::Error &error) { Warning() << "Failed to read current value from storage: " << error.message; });
725 return current;
726}
727
728void GenericResource::createOrModify(Sink::Storage::Transaction &transaction, Sink::Storage::Transaction &synchronizationTransaction,
729 DomainTypeAdaptorFactoryInterface &adaptorFactory, const QByteArray &bufferType, const QByteArray &remoteId, const Sink::ApplicationDomain::ApplicationDomainType &entity)
730{ 662{
731 auto mainDatabase = Storage::mainDatabase(transaction, bufferType); 663 Trace() << "Create or modify" << bufferType << remoteId;
732 const auto sinkId = resolveRemoteId(bufferType, remoteId, synchronizationTransaction); 664 auto mainDatabase = Storage::mainDatabase(mTransaction, bufferType);
665 const auto sinkId = syncStore().resolveRemoteId(bufferType, remoteId);
733 const auto found = mainDatabase.contains(sinkId); 666 const auto found = mainDatabase.contains(sinkId);
667 auto adaptorFactory = Sink::AdaptorFactoryRegistry::instance().getFactory(mResourceType, bufferType);
734 if (!found) { 668 if (!found) {
735 Trace() << "Found a new entity: " << remoteId; 669 Trace() << "Found a new entity: " << remoteId;
736 createEntity( 670 createEntity(
737 sinkId, bufferType, entity, adaptorFactory, [this](const QByteArray &buffer) { enqueueCommand(mSynchronizerQueue, Sink::Commands::CreateEntityCommand, buffer); }); 671 sinkId, bufferType, entity, *adaptorFactory, [this](const QByteArray &buffer) { enqueueCommand(Sink::Commands::CreateEntityCommand, buffer); });
738 } else { // modification 672 } else { // modification
739 if (auto current = getLatest(mainDatabase, sinkId, adaptorFactory)) { 673 if (auto current = store().getLatest(mainDatabase, sinkId, *adaptorFactory)) {
740 bool changed = false; 674 bool changed = false;
741 for (const auto &property : entity.changedProperties()) { 675 for (const auto &property : entity.changedProperties()) {
742 if (entity.getProperty(property) != current->getProperty(property)) { 676 if (entity.getProperty(property) != current->getProperty(property)) {
@@ -746,8 +680,8 @@ void GenericResource::createOrModify(Sink::Storage::Transaction &transaction, Si
746 } 680 }
747 if (changed) { 681 if (changed) {
748 Trace() << "Found a modified entity: " << remoteId; 682 Trace() << "Found a modified entity: " << remoteId;
749 modifyEntity(sinkId, Sink::Storage::maxRevision(transaction), bufferType, entity, adaptorFactory, 683 modifyEntity(sinkId, Sink::Storage::maxRevision(mTransaction), bufferType, entity, *adaptorFactory,
750 [this](const QByteArray &buffer) { enqueueCommand(mSynchronizerQueue, Sink::Commands::ModifyEntityCommand, buffer); }); 684 [this](const QByteArray &buffer) { enqueueCommand(Sink::Commands::ModifyEntityCommand, buffer); });
751 } 685 }
752 } else { 686 } else {
753 Warning() << "Failed to get current entity"; 687 Warning() << "Failed to get current entity";
@@ -755,6 +689,118 @@ void GenericResource::createOrModify(Sink::Storage::Transaction &transaction, Si
755 } 689 }
756} 690}
757 691
692KAsync::Job<void> Synchronizer::synchronize()
693{
694 mTransaction = mStorage.createTransaction(Sink::Storage::ReadOnly);
695 mSyncTransaction = mSyncStorage.createTransaction(Sink::Storage::ReadWrite);
696 return synchronizeWithSource().then<void>([this]() {
697 mTransaction.abort();
698 mSyncTransaction.commit();
699 mSyncStore.clear();
700 mEntityStore.clear();
701 });
702}
703
704
705
706SourceWriteBack::SourceWriteBack(const QByteArray &resourceType, const QByteArray &resourceInstanceIdentifier)
707 : ChangeReplay(resourceInstanceIdentifier),
708 mSyncStorage(Sink::storageLocation(), resourceInstanceIdentifier + ".synchronization", Sink::Storage::ReadWrite),
709 mResourceType(resourceType),
710 mResourceInstanceIdentifier(resourceInstanceIdentifier)
711{
712
713}
714
715EntityStore &SourceWriteBack::store()
716{
717 if (!mEntityStore) {
718 mEntityStore = QSharedPointer<EntityStore>::create(mResourceType, mResourceInstanceIdentifier, mTransaction);
719 }
720 return *mEntityStore;
721}
722
723SyncStore &SourceWriteBack::syncStore()
724{
725 if (!mSyncStore) {
726 mSyncStore = QSharedPointer<SyncStore>::create(mSyncTransaction);
727 }
728 return *mSyncStore;
729}
730
731KAsync::Job<void> SourceWriteBack::replay(const QByteArray &type, const QByteArray &key, const QByteArray &value)
732{
733 mTransaction = mStorage.createTransaction(Sink::Storage::ReadOnly);
734 mSyncTransaction = mSyncStorage.createTransaction(Sink::Storage::ReadWrite);
735
736 Sink::EntityBuffer buffer(value);
737 const Sink::Entity &entity = buffer.entity();
738 const auto metadataBuffer = Sink::EntityBuffer::readBuffer<Sink::Metadata>(entity.metadata());
739 Q_ASSERT(metadataBuffer);
740 if (!metadataBuffer->replayToSource()) {
741 Trace() << "Change is coming from the source";
742 return KAsync::null<void>();
743 }
744 const qint64 revision = metadataBuffer ? metadataBuffer->revision() : -1;
745 const auto operation = metadataBuffer ? metadataBuffer->operation() : Sink::Operation_Creation;
746 const auto uid = Sink::Storage::uidFromKey(key);
747 QByteArray oldRemoteId;
748
749 if (operation != Sink::Operation_Creation) {
750 oldRemoteId = syncStore().resolveLocalId(type, uid);
751 }
752 Trace() << "Replaying " << key << type;
753
754 KAsync::Job<QByteArray> job = KAsync::null<QByteArray>();
755 if (type == ENTITY_TYPE_FOLDER) {
756 auto folder = store().read<ApplicationDomain::Folder>(uid);
757 // const Sink::ApplicationDomain::Folder folder(mResourceInstanceIdentifier, uid, revision, mAdaptorFactories.value(type)->createAdaptor(entity));
758 job = replay(folder, operation, oldRemoteId);
759 } else if (type == ENTITY_TYPE_MAIL) {
760 auto mail = store().read<ApplicationDomain::Mail>(uid);
761 // const Sink::ApplicationDomain::Mail mail(mResourceInstanceIdentifier, uid, revision, mAdaptorFactories.value(type)->createAdaptor(entity));
762 job = replay(mail, operation, oldRemoteId);
763 }
764
765 return job.then<void, QByteArray>([this, operation, type, uid](const QByteArray &remoteId) {
766 Trace() << "Replayed change with remote id: " << remoteId;
767 if (operation == Sink::Operation_Creation) {
768 if (remoteId.isEmpty()) {
769 Warning() << "Returned an empty remoteId from the creation";
770 } else {
771 syncStore().recordRemoteId(type, uid, remoteId);
772 }
773 } else if (operation == Sink::Operation_Modification) {
774 if (remoteId.isEmpty()) {
775 Warning() << "Returned an empty remoteId from the creation";
776 } else {
777 syncStore().updateRemoteId(type, uid, remoteId);
778 }
779 } else if (operation == Sink::Operation_Removal) {
780 syncStore().removeRemoteId(type, uid, remoteId);
781 } else {
782 Warning() << "Unkown operation" << operation;
783 }
784
785 mTransaction.abort();
786 mSyncTransaction.commit();
787 mSyncStore.clear();
788 mEntityStore.clear();
789 }, [](int errorCode, const QString &errorMessage) {
790 Warning() << "Failed to replay change: " << errorMessage;
791 });
792}
793
794KAsync::Job<QByteArray> SourceWriteBack::replay(const ApplicationDomain::Mail &, Sink::Operation, const QByteArray &)
795{
796 return KAsync::null<QByteArray>();
797}
798
799KAsync::Job<QByteArray> SourceWriteBack::replay(const ApplicationDomain::Folder &, Sink::Operation, const QByteArray &)
800{
801 return KAsync::null<QByteArray>();
802}
803
758 804
759#pragma clang diagnostic push 805#pragma clang diagnostic push
760#pragma clang diagnostic ignored "-Wundefined-reinterpret-cast" 806#pragma clang diagnostic ignored "-Wundefined-reinterpret-cast"
diff --git a/common/genericresource.h b/common/genericresource.h
index c551e29..45d5d3a 100644
--- a/common/genericresource.h
+++ b/common/genericresource.h
@@ -24,14 +24,16 @@
24#include <messagequeue.h> 24#include <messagequeue.h>
25#include <flatbuffers/flatbuffers.h> 25#include <flatbuffers/flatbuffers.h>
26#include <domainadaptor.h> 26#include <domainadaptor.h>
27#include "changereplay.h"
28
27#include <QTimer> 29#include <QTimer>
28 30
29class CommandProcessor; 31class CommandProcessor;
30class ChangeReplay;
31 32
32namespace Sink { 33namespace Sink {
33class Pipeline; 34class Pipeline;
34class Preprocessor; 35class Preprocessor;
36class Synchronizer;
35 37
36/** 38/**
37 * Generic Resource implementation. 39 * Generic Resource implementation.
@@ -39,7 +41,7 @@ class Preprocessor;
39class SINK_EXPORT GenericResource : public Resource 41class SINK_EXPORT GenericResource : public Resource
40{ 42{
41public: 43public:
42 GenericResource(const QByteArray &resourceInstanceIdentifier, const QSharedPointer<Pipeline> &pipeline = QSharedPointer<Pipeline>()); 44 GenericResource(const QByteArray &resourceType, const QByteArray &resourceInstanceIdentifier, const QSharedPointer<Pipeline> &pipeline, const QSharedPointer<ChangeReplay> &changeReplay, const QSharedPointer<Synchronizer> &synchronizer);
43 virtual ~GenericResource(); 45 virtual ~GenericResource();
44 46
45 virtual void processCommand(int commandId, const QByteArray &data) Q_DECL_OVERRIDE; 47 virtual void processCommand(int commandId, const QByteArray &data) Q_DECL_OVERRIDE;
@@ -64,41 +66,90 @@ protected:
64 66
65 void addType(const QByteArray &type, DomainTypeAdaptorFactoryInterface::Ptr factory, const QVector<Sink::Preprocessor *> &preprocessors); 67 void addType(const QByteArray &type, DomainTypeAdaptorFactoryInterface::Ptr factory, const QVector<Sink::Preprocessor *> &preprocessors);
66 68
67 ///Base implementation call the replay$Type calls
68 virtual KAsync::Job<void> replay(Sink::Storage &synchronizationStore, const QByteArray &type, const QByteArray &key, const QByteArray &value);
69 ///Implement to write back changes to the server
70 virtual KAsync::Job<QByteArray> replay(const Sink::ApplicationDomain::Mail &, Sink::Operation, const QByteArray &oldRemoteId);
71 virtual KAsync::Job<QByteArray> replay(const Sink::ApplicationDomain::Folder &, Sink::Operation, const QByteArray &oldRemoteId);
72
73 void onProcessorError(int errorCode, const QString &errorMessage); 69 void onProcessorError(int errorCode, const QString &errorMessage);
74 void enqueueCommand(MessageQueue &mq, int commandId, const QByteArray &data); 70 void enqueueCommand(MessageQueue &mq, int commandId, const QByteArray &data);
75 71
76 static void createEntity(const QByteArray &localId, const QByteArray &bufferType, const Sink::ApplicationDomain::ApplicationDomainType &domainObject, 72 MessageQueue mUserQueue;
77 DomainTypeAdaptorFactoryInterface &adaptorFactory, std::function<void(const QByteArray &)> callback); 73 MessageQueue mSynchronizerQueue;
78 static void modifyEntity(const QByteArray &localId, qint64 revision, const QByteArray &bufferType, const Sink::ApplicationDomain::ApplicationDomainType &domainObject, 74 QByteArray mResourceType;
79 DomainTypeAdaptorFactoryInterface &adaptorFactory, std::function<void(const QByteArray &)> callback); 75 QByteArray mResourceInstanceIdentifier;
80 static void deleteEntity(const QByteArray &localId, qint64 revision, const QByteArray &bufferType, std::function<void(const QByteArray &)> callback); 76 QSharedPointer<Pipeline> mPipeline;
77
78private:
79 CommandProcessor *mProcessor;
80 QSharedPointer<ChangeReplay> mChangeReplay;
81 QSharedPointer<Synchronizer> mSynchronizer;
82 int mError;
83 QTimer mCommitQueueTimer;
84 qint64 mClientLowerBoundRevision;
85 QHash<QByteArray, DomainTypeAdaptorFactoryInterface::Ptr> mAdaptorFactories;
86};
87
88class SINK_EXPORT SyncStore
89{
90public:
91 SyncStore(Sink::Storage::Transaction &);
81 92
82 /** 93 /**
83 * Records a localId to remoteId mapping 94 * Records a localId to remoteId mapping
84 */ 95 */
85 void recordRemoteId(const QByteArray &bufferType, const QByteArray &localId, const QByteArray &remoteId, Sink::Storage::Transaction &transaction); 96 void recordRemoteId(const QByteArray &bufferType, const QByteArray &localId, const QByteArray &remoteId);
86 void removeRemoteId(const QByteArray &bufferType, const QByteArray &localId, const QByteArray &remoteId, Sink::Storage::Transaction &transaction); 97 void removeRemoteId(const QByteArray &bufferType, const QByteArray &localId, const QByteArray &remoteId);
87 void updateRemoteId(const QByteArray &bufferType, const QByteArray &localId, const QByteArray &remoteId, Sink::Storage::Transaction &transaction); 98 void updateRemoteId(const QByteArray &bufferType, const QByteArray &localId, const QByteArray &remoteId);
88 99
89 /** 100 /**
90 * Tries to find a local id for the remote id, and creates a new local id otherwise. 101 * Tries to find a local id for the remote id, and creates a new local id otherwise.
91 * 102 *
92 * The new local id is recorded in the local to remote id mapping. 103 * The new local id is recorded in the local to remote id mapping.
93 */ 104 */
94 QByteArray resolveRemoteId(const QByteArray &type, const QByteArray &remoteId, Sink::Storage::Transaction &transaction); 105 QByteArray resolveRemoteId(const QByteArray &type, const QByteArray &remoteId);
95 106
96 /** 107 /**
97 * Tries to find a remote id for a local id. 108 * Tries to find a remote id for a local id.
98 * 109 *
99 * This can fail if the entity hasn't been written back to the server yet. 110 * This can fail if the entity hasn't been written back to the server yet.
100 */ 111 */
101 QByteArray resolveLocalId(const QByteArray &bufferType, const QByteArray &localId, Sink::Storage::Transaction &transaction); 112 QByteArray resolveLocalId(const QByteArray &bufferType, const QByteArray &localId);
113
114private:
115 Sink::Storage::Transaction &mTransaction;
116};
117
118class SINK_EXPORT EntityStore
119{
120public:
121 EntityStore(const QByteArray &resourceType, const QByteArray &mResourceInstanceIdentifier, Sink::Storage::Transaction &transaction);
122
123 template<typename T>
124 T read(const QByteArray &identifier) const;
125
126 static QSharedPointer<Sink::ApplicationDomain::BufferAdaptor> getLatest(const Sink::Storage::NamedDatabase &db, const QByteArray &uid, DomainTypeAdaptorFactoryInterface &adaptorFactory);
127private:
128 QByteArray mResourceType;
129 QByteArray mResourceInstanceIdentifier;
130 Sink::Storage::Transaction &mTransaction;
131};
132
133/**
134 * Synchronize and add what we don't already have to local queue
135 */
136class SINK_EXPORT Synchronizer
137{
138public:
139 Synchronizer(const QByteArray &resourceType, const QByteArray &resourceInstanceIdentifier);
140
141 void setup(const std::function<void(int commandId, const QByteArray &data)> &enqueueCommandCallback);
142 KAsync::Job<void> synchronize();
143
144protected:
145 ///Calls the callback to enqueue the command
146 void enqueueCommand(int commandId, const QByteArray &data);
147
148 static void createEntity(const QByteArray &localId, const QByteArray &bufferType, const Sink::ApplicationDomain::ApplicationDomainType &domainObject,
149 DomainTypeAdaptorFactoryInterface &adaptorFactory, std::function<void(const QByteArray &)> callback);
150 static void modifyEntity(const QByteArray &localId, qint64 revision, const QByteArray &bufferType, const Sink::ApplicationDomain::ApplicationDomainType &domainObject,
151 DomainTypeAdaptorFactoryInterface &adaptorFactory, std::function<void(const QByteArray &)> callback);
152 static void deleteEntity(const QByteArray &localId, qint64 revision, const QByteArray &bufferType, std::function<void(const QByteArray &)> callback);
102 153
103 /** 154 /**
104 * A synchronous algorithm to remove entities that are no longer existing. 155 * A synchronous algorithm to remove entities that are no longer existing.
@@ -110,7 +161,7 @@ protected:
110 * 161 *
111 * All functions are called synchronously, and both @param entryGenerator and @param exists need to be synchronous. 162 * All functions are called synchronously, and both @param entryGenerator and @param exists need to be synchronous.
112 */ 163 */
113 void scanForRemovals(Sink::Storage::Transaction &transaction, Sink::Storage::Transaction &synchronizationTransaction, const QByteArray &bufferType, 164 void scanForRemovals(const QByteArray &bufferType,
114 const std::function<void(const std::function<void(const QByteArray &key)> &callback)> &entryGenerator, std::function<bool(const QByteArray &remoteId)> exists); 165 const std::function<void(const std::function<void(const QByteArray &key)> &callback)> &entryGenerator, std::function<bool(const QByteArray &remoteId)> exists);
115 166
116 /** 167 /**
@@ -118,22 +169,60 @@ protected:
118 * 169 *
119 * Depending on whether the entity is locally available, or has changed. 170 * Depending on whether the entity is locally available, or has changed.
120 */ 171 */
121 void createOrModify(Sink::Storage::Transaction &transaction, Sink::Storage::Transaction &synchronizationTransaction, DomainTypeAdaptorFactoryInterface &adaptorFactory, 172 void createOrModify(const QByteArray &bufferType, const QByteArray &remoteId, const Sink::ApplicationDomain::ApplicationDomainType &entity);
122 const QByteArray &bufferType, const QByteArray &remoteId, const Sink::ApplicationDomain::ApplicationDomainType &entity);
123 173
124 static QSharedPointer<Sink::ApplicationDomain::BufferAdaptor> getLatest(const Sink::Storage::NamedDatabase &db, const QByteArray &uid, DomainTypeAdaptorFactoryInterface &adaptorFactory); 174 //Read only access to main storage
175 EntityStore &store();
125 176
126 MessageQueue mUserQueue; 177 //Read/Write access to sync storage
127 MessageQueue mSynchronizerQueue; 178 SyncStore &syncStore();
179
180 virtual KAsync::Job<void> synchronizeWithSource() = 0;
181
182private:
183 QSharedPointer<SyncStore> mSyncStore;
184 QSharedPointer<EntityStore> mEntityStore;
185 Sink::Storage mStorage;
186 Sink::Storage mSyncStorage;
187 QByteArray mResourceType;
128 QByteArray mResourceInstanceIdentifier; 188 QByteArray mResourceInstanceIdentifier;
129 QSharedPointer<Pipeline> mPipeline; 189 Sink::Storage::Transaction mTransaction;
190 Sink::Storage::Transaction mSyncTransaction;
191 std::function<void(int commandId, const QByteArray &data)> mEnqueue;
192};
193
194/**
195 * Replay changes to the source
196 */
197class SINK_EXPORT SourceWriteBack : public ChangeReplay
198{
199public:
200 SourceWriteBack(const QByteArray &resourceType,const QByteArray &resourceInstanceIdentifier);
201
202protected:
203 ///Base implementation calls the replay$Type calls
204 virtual KAsync::Job<void> replay(const QByteArray &type, const QByteArray &key, const QByteArray &value) Q_DECL_OVERRIDE;
205
206protected:
207 ///Implement to write back changes to the server
208 virtual KAsync::Job<QByteArray> replay(const Sink::ApplicationDomain::Mail &, Sink::Operation, const QByteArray &oldRemoteId);
209 virtual KAsync::Job<QByteArray> replay(const Sink::ApplicationDomain::Folder &, Sink::Operation, const QByteArray &oldRemoteId);
210
211 //Read only access to main storage
212 EntityStore &store();
213
214 //Read/Write access to sync storage
215 SyncStore &syncStore();
130 216
131private: 217private:
132 CommandProcessor *mProcessor; 218 Sink::Storage mSyncStorage;
133 ChangeReplay *mSourceChangeReplay; 219 QSharedPointer<SyncStore> mSyncStore;
134 int mError; 220 QSharedPointer<EntityStore> mEntityStore;
135 QTimer mCommitQueueTimer; 221 Sink::Storage::Transaction mTransaction;
136 qint64 mClientLowerBoundRevision; 222 Sink::Storage::Transaction mSyncTransaction;
137 QHash<QByteArray, DomainTypeAdaptorFactoryInterface::Ptr> mAdaptorFactories; 223 QByteArray mResourceType;
224 QByteArray mResourceInstanceIdentifier;
138}; 225};
226
227
139} 228}
diff --git a/common/pipeline.cpp b/common/pipeline.cpp
index 637a1b8..7863f67 100644
--- a/common/pipeline.cpp
+++ b/common/pipeline.cpp
@@ -34,6 +34,7 @@
34#include "entitybuffer.h" 34#include "entitybuffer.h"
35#include "log.h" 35#include "log.h"
36#include "domain/applicationdomaintype.h" 36#include "domain/applicationdomaintype.h"
37#include "adaptorfactoryregistry.h"
37#include "definitions.h" 38#include "definitions.h"
38#include "bufferutils.h" 39#include "bufferutils.h"
39 40
@@ -52,11 +53,11 @@ public:
52 Storage storage; 53 Storage storage;
53 Storage::Transaction transaction; 54 Storage::Transaction transaction;
54 QHash<QString, QVector<Preprocessor *>> processors; 55 QHash<QString, QVector<Preprocessor *>> processors;
55 QHash<QString, DomainTypeAdaptorFactoryInterface::Ptr> adaptorFactory;
56 bool revisionChanged; 56 bool revisionChanged;
57 void storeNewRevision(qint64 newRevision, const flatbuffers::FlatBufferBuilder &fbb, const QByteArray &bufferType, const QByteArray &uid); 57 void storeNewRevision(qint64 newRevision, const flatbuffers::FlatBufferBuilder &fbb, const QByteArray &bufferType, const QByteArray &uid);
58 QTime transactionTime; 58 QTime transactionTime;
59 int transactionItemCount; 59 int transactionItemCount;
60 QByteArray resourceType;
60}; 61};
61 62
62void Pipeline::Private::storeNewRevision(qint64 newRevision, const flatbuffers::FlatBufferBuilder &fbb, const QByteArray &bufferType, const QByteArray &uid) 63void Pipeline::Private::storeNewRevision(qint64 newRevision, const flatbuffers::FlatBufferBuilder &fbb, const QByteArray &bufferType, const QByteArray &uid)
@@ -84,9 +85,9 @@ void Pipeline::setPreprocessors(const QString &entityType, const QVector<Preproc
84 d->processors[entityType] = processors; 85 d->processors[entityType] = processors;
85} 86}
86 87
87void Pipeline::setAdaptorFactory(const QString &entityType, DomainTypeAdaptorFactoryInterface::Ptr factory) 88void Pipeline::setResourceType(const QByteArray &resourceType)
88{ 89{
89 d->adaptorFactory.insert(entityType, factory); 90 d->resourceType = resourceType;
90} 91}
91 92
92void Pipeline::startTransaction() 93void Pipeline::startTransaction()
@@ -102,7 +103,9 @@ void Pipeline::startTransaction()
102 Trace() << "Starting transaction."; 103 Trace() << "Starting transaction.";
103 d->transactionTime.start(); 104 d->transactionTime.start();
104 d->transactionItemCount = 0; 105 d->transactionItemCount = 0;
105 d->transaction = std::move(storage().createTransaction(Storage::ReadWrite)); 106 d->transaction = std::move(storage().createTransaction(Storage::ReadWrite, [](const Sink::Storage::Error &error) {
107 Warning() << error.message;
108 }));
106} 109}
107 110
108void Pipeline::commit() 111void Pipeline::commit()
@@ -189,7 +192,7 @@ KAsync::Job<qint64> Pipeline::newEntity(void const *command, size_t size)
189 auto metadataBuffer = metadataBuilder.Finish(); 192 auto metadataBuffer = metadataBuilder.Finish();
190 FinishMetadataBuffer(metadataFbb, metadataBuffer); 193 FinishMetadataBuffer(metadataFbb, metadataBuffer);
191 194
192 auto adaptorFactory = d->adaptorFactory.value(bufferType); 195 auto adaptorFactory = Sink::AdaptorFactoryRegistry::instance().getFactory(d->resourceType, bufferType);
193 if (!adaptorFactory) { 196 if (!adaptorFactory) {
194 Warning() << "no adaptor factory for type " << bufferType; 197 Warning() << "no adaptor factory for type " << bufferType;
195 return KAsync::error<qint64>(0); 198 return KAsync::error<qint64>(0);
@@ -244,7 +247,7 @@ KAsync::Job<qint64> Pipeline::modifiedEntity(void const *command, size_t size)
244 } 247 }
245 248
246 // TODO use only readPropertyMapper and writePropertyMapper 249 // TODO use only readPropertyMapper and writePropertyMapper
247 auto adaptorFactory = d->adaptorFactory.value(bufferType); 250 auto adaptorFactory = Sink::AdaptorFactoryRegistry::instance().getFactory(d->resourceType, bufferType);
248 if (!adaptorFactory) { 251 if (!adaptorFactory) {
249 Warning() << "no adaptor factory for type " << bufferType; 252 Warning() << "no adaptor factory for type " << bufferType;
250 return KAsync::error<qint64>(0); 253 return KAsync::error<qint64>(0);
@@ -373,7 +376,7 @@ KAsync::Job<qint64> Pipeline::deletedEntity(void const *command, size_t size)
373 flatbuffers::FlatBufferBuilder fbb; 376 flatbuffers::FlatBufferBuilder fbb;
374 EntityBuffer::assembleEntityBuffer(fbb, metadataFbb.GetBufferPointer(), metadataFbb.GetSize(), 0, 0, 0, 0); 377 EntityBuffer::assembleEntityBuffer(fbb, metadataFbb.GetBufferPointer(), metadataFbb.GetSize(), 0, 0, 0, 0);
375 378
376 auto adaptorFactory = d->adaptorFactory.value(bufferType); 379 auto adaptorFactory = Sink::AdaptorFactoryRegistry::instance().getFactory(d->resourceType, bufferType);
377 if (!adaptorFactory) { 380 if (!adaptorFactory) {
378 Warning() << "no adaptor factory for type " << bufferType; 381 Warning() << "no adaptor factory for type " << bufferType;
379 return KAsync::error<qint64>(0); 382 return KAsync::error<qint64>(0);
diff --git a/common/pipeline.h b/common/pipeline.h
index c65cbfd..2ca87a4 100644
--- a/common/pipeline.h
+++ b/common/pipeline.h
@@ -46,13 +46,12 @@ public:
46 46
47 Storage &storage() const; 47 Storage &storage() const;
48 48
49 void setResourceType(const QByteArray &resourceType);
49 void setPreprocessors(const QString &entityType, const QVector<Preprocessor *> &preprocessors); 50 void setPreprocessors(const QString &entityType, const QVector<Preprocessor *> &preprocessors);
50 void startTransaction(); 51 void startTransaction();
51 void commit(); 52 void commit();
52 Storage::Transaction &transaction(); 53 Storage::Transaction &transaction();
53 54
54 void setAdaptorFactory(const QString &entityType, DomainTypeAdaptorFactoryInterface::Ptr factory);
55
56 KAsync::Job<qint64> newEntity(void const *command, size_t size); 55 KAsync::Job<qint64> newEntity(void const *command, size_t size);
57 KAsync::Job<qint64> modifiedEntity(void const *command, size_t size); 56 KAsync::Job<qint64> modifiedEntity(void const *command, size_t size);
58 KAsync::Job<qint64> deletedEntity(void const *command, size_t size); 57 KAsync::Job<qint64> deletedEntity(void const *command, size_t size);
diff --git a/common/resource.cpp b/common/resource.cpp
index 6713686..82c9fc8 100644
--- a/common/resource.cpp
+++ b/common/resource.cpp
@@ -26,6 +26,7 @@
26#include <QPointer> 26#include <QPointer>
27 27
28#include "facadefactory.h" 28#include "facadefactory.h"
29#include "adaptorfactoryregistry.h"
29 30
30namespace Sink { 31namespace Sink {
31 32
@@ -110,6 +111,7 @@ ResourceFactory *ResourceFactory::load(const QString &resourceName)
110 if (factory) { 111 if (factory) {
111 Private::s_loadedFactories.insert(resourceName, factory); 112 Private::s_loadedFactories.insert(resourceName, factory);
112 factory->registerFacades(FacadeFactory::instance()); 113 factory->registerFacades(FacadeFactory::instance());
114 factory->registerAdaptorFactories(AdaptorFactoryRegistry::instance());
113 // TODO: if we need more data on it const QJsonObject json = loader.metaData()[QStringLiteral("MetaData")].toObject(); 115 // TODO: if we need more data on it const QJsonObject json = loader.metaData()[QStringLiteral("MetaData")].toObject();
114 return factory; 116 return factory;
115 } else { 117 } else {
diff --git a/common/resource.h b/common/resource.h
index 0e7cd11..d6c3c5f 100644
--- a/common/resource.h
+++ b/common/resource.h
@@ -26,6 +26,7 @@
26 26
27namespace Sink { 27namespace Sink {
28class FacadeFactory; 28class FacadeFactory;
29class AdaptorFactoryRegistry;
29 30
30/** 31/**
31 * Resource interface 32 * Resource interface
@@ -81,6 +82,7 @@ public:
81 82
82 virtual Resource *createResource(const QByteArray &instanceIdentifier) = 0; 83 virtual Resource *createResource(const QByteArray &instanceIdentifier) = 0;
83 virtual void registerFacades(FacadeFactory &factory) = 0; 84 virtual void registerFacades(FacadeFactory &factory) = 0;
85 virtual void registerAdaptorFactories(AdaptorFactoryRegistry &registry) {};
84 86
85private: 87private:
86 class Private; 88 class Private;
diff --git a/examples/dummyresource/resourcefactory.cpp b/examples/dummyresource/resourcefactory.cpp
index 48858da..609d23e 100644
--- a/examples/dummyresource/resourcefactory.cpp
+++ b/examples/dummyresource/resourcefactory.cpp
@@ -35,6 +35,7 @@
35#include "definitions.h" 35#include "definitions.h"
36#include "facadefactory.h" 36#include "facadefactory.h"
37#include "indexupdater.h" 37#include "indexupdater.h"
38#include "adaptorfactoryregistry.h"
38#include <QDate> 39#include <QDate>
39#include <QUuid> 40#include <QUuid>
40 41
@@ -43,8 +44,87 @@
43#define ENTITY_TYPE_MAIL "mail" 44#define ENTITY_TYPE_MAIL "mail"
44#define ENTITY_TYPE_FOLDER "folder" 45#define ENTITY_TYPE_FOLDER "folder"
45 46
47class DummySynchronizer : public Sink::Synchronizer {
48 public:
49
50 DummySynchronizer(const QByteArray &resourceType, const QByteArray &resourceInstanceIdentifier)
51 : Sink::Synchronizer(resourceType, resourceInstanceIdentifier)
52 {
53
54 }
55
56 Sink::ApplicationDomain::Event::Ptr createEvent(const QByteArray &ridBuffer, const QMap<QString, QVariant> &data)
57 {
58 static uint8_t rawData[100];
59 auto event = Sink::ApplicationDomain::Event::Ptr::create();
60 event->setProperty("summary", data.value("summary").toString());
61 event->setProperty("remoteId", ridBuffer);
62 event->setProperty("description", data.value("description").toString());
63 event->setProperty("attachment", QByteArray::fromRawData(reinterpret_cast<const char*>(rawData), 100));
64 return event;
65 }
66
67 Sink::ApplicationDomain::Mail::Ptr createMail(const QByteArray &ridBuffer, const QMap<QString, QVariant> &data)
68 {
69 auto mail = Sink::ApplicationDomain::Mail::Ptr::create();
70 mail->setProperty("subject", data.value("subject").toString());
71 mail->setProperty("senderEmail", data.value("senderEmail").toString());
72 mail->setProperty("senderName", data.value("senderName").toString());
73 mail->setProperty("date", data.value("date").toString());
74 mail->setProperty("folder", syncStore().resolveRemoteId(ENTITY_TYPE_FOLDER, data.value("parentFolder").toByteArray()));
75 mail->setProperty("unread", data.value("unread").toBool());
76 mail->setProperty("important", data.value("important").toBool());
77 return mail;
78 }
79
80 Sink::ApplicationDomain::Folder::Ptr createFolder(const QByteArray &ridBuffer, const QMap<QString, QVariant> &data)
81 {
82 auto folder = Sink::ApplicationDomain::Folder::Ptr::create();
83 folder->setProperty("name", data.value("name").toString());
84 folder->setProperty("icon", data.value("icon").toString());
85 if (!data.value("parent").toString().isEmpty()) {
86 auto sinkId = syncStore().resolveRemoteId(ENTITY_TYPE_FOLDER, data.value("parent").toByteArray());
87 folder->setProperty("parent", sinkId);
88 }
89 return folder;
90 }
91
92 void synchronize(const QByteArray &bufferType, const QMap<QString, QMap<QString, QVariant> > &data, std::function<Sink::ApplicationDomain::ApplicationDomainType::Ptr(const QByteArray &ridBuffer, const QMap<QString, QVariant> &data)> createEntity)
93 {
94 auto time = QSharedPointer<QTime>::create();
95 time->start();
96 //TODO find items to remove
97 int count = 0;
98 for (auto it = data.constBegin(); it != data.constEnd(); it++) {
99 count++;
100 const auto remoteId = it.key().toUtf8();
101 auto entity = createEntity(remoteId, it.value());
102 createOrModify(bufferType, remoteId, *entity);
103 }
104 Trace() << "Sync of " << count << " entities of type " << bufferType << " done." << Sink::Log::TraceTime(time->elapsed());
105 }
106
107 KAsync::Job<void> synchronizeWithSource() Q_DECL_OVERRIDE
108 {
109 Log() << " Synchronizing with the source";
110 return KAsync::start<void>([this]() {
111 synchronize(ENTITY_TYPE_EVENT, DummyStore::instance().events(), [this](const QByteArray &ridBuffer, const QMap<QString, QVariant> &data) {
112 return createEvent(ridBuffer, data);
113 });
114 synchronize(ENTITY_TYPE_MAIL, DummyStore::instance().mails(), [this](const QByteArray &ridBuffer, const QMap<QString, QVariant> &data) {
115 return createMail(ridBuffer, data);
116 });
117 synchronize(ENTITY_TYPE_FOLDER, DummyStore::instance().folders(), [this](const QByteArray &ridBuffer, const QMap<QString, QVariant> &data) {
118 return createFolder(ridBuffer, data);
119 });
120 Log() << "Done Synchronizing";
121 });
122 }
123
124};
125
46DummyResource::DummyResource(const QByteArray &instanceIdentifier, const QSharedPointer<Sink::Pipeline> &pipeline) 126DummyResource::DummyResource(const QByteArray &instanceIdentifier, const QSharedPointer<Sink::Pipeline> &pipeline)
47 : Sink::GenericResource(instanceIdentifier, pipeline), 127 : Sink::GenericResource(PLUGIN_NAME, instanceIdentifier, pipeline, QSharedPointer<Sink::NullChangeReplay>::create(), QSharedPointer<DummySynchronizer>::create(PLUGIN_NAME, instanceIdentifier)),
48 mEventAdaptorFactory(QSharedPointer<DummyEventAdaptorFactory>::create()), 128 mEventAdaptorFactory(QSharedPointer<DummyEventAdaptorFactory>::create()),
49 mMailAdaptorFactory(QSharedPointer<DummyMailAdaptorFactory>::create()), 129 mMailAdaptorFactory(QSharedPointer<DummyMailAdaptorFactory>::create()),
50 mFolderAdaptorFactory(QSharedPointer<DummyFolderAdaptorFactory>::create()) 130 mFolderAdaptorFactory(QSharedPointer<DummyFolderAdaptorFactory>::create())
@@ -57,80 +137,9 @@ DummyResource::DummyResource(const QByteArray &instanceIdentifier, const QShared
57 QVector<Sink::Preprocessor*>() << new DefaultIndexUpdater<Sink::ApplicationDomain::Event>); 137 QVector<Sink::Preprocessor*>() << new DefaultIndexUpdater<Sink::ApplicationDomain::Event>);
58} 138}
59 139
60Sink::ApplicationDomain::Event::Ptr DummyResource::createEvent(const QByteArray &ridBuffer, const QMap<QString, QVariant> &data, Sink::Storage::Transaction &transaction) 140DummyResource::~DummyResource()
61{
62 static uint8_t rawData[100];
63 auto event = Sink::ApplicationDomain::Event::Ptr::create();
64 event->setProperty("summary", data.value("summary").toString());
65 event->setProperty("remoteId", ridBuffer);
66 event->setProperty("description", data.value("description").toString());
67 event->setProperty("attachment", QByteArray::fromRawData(reinterpret_cast<const char*>(rawData), 100));
68 return event;
69}
70
71Sink::ApplicationDomain::Mail::Ptr DummyResource::createMail(const QByteArray &ridBuffer, const QMap<QString, QVariant> &data, Sink::Storage::Transaction &transaction)
72{ 141{
73 auto mail = Sink::ApplicationDomain::Mail::Ptr::create();
74 mail->setProperty("subject", data.value("subject").toString());
75 mail->setProperty("senderEmail", data.value("senderEmail").toString());
76 mail->setProperty("senderName", data.value("senderName").toString());
77 mail->setProperty("date", data.value("date").toString());
78 mail->setProperty("folder", resolveRemoteId(ENTITY_TYPE_FOLDER, data.value("parentFolder").toByteArray(), transaction));
79 mail->setProperty("unread", data.value("unread").toBool());
80 mail->setProperty("important", data.value("important").toBool());
81 return mail;
82}
83 142
84Sink::ApplicationDomain::Folder::Ptr DummyResource::createFolder(const QByteArray &ridBuffer, const QMap<QString, QVariant> &data, Sink::Storage::Transaction &transaction)
85{
86 auto folder = Sink::ApplicationDomain::Folder::Ptr::create();
87 folder->setProperty("name", data.value("name").toString());
88 folder->setProperty("icon", data.value("icon").toString());
89 if (!data.value("parent").toString().isEmpty()) {
90 auto sinkId = resolveRemoteId(ENTITY_TYPE_FOLDER, data.value("parent").toByteArray(), transaction);
91 folder->setProperty("parent", sinkId);
92 }
93 return folder;
94}
95
96void DummyResource::synchronize(const QByteArray &bufferType, const QMap<QString, QMap<QString, QVariant> > &data, Sink::Storage::Transaction &transaction, Sink::Storage::Transaction &synchronizationTransaction, DomainTypeAdaptorFactoryInterface &adaptorFactory, std::function<Sink::ApplicationDomain::ApplicationDomainType::Ptr(const QByteArray &ridBuffer, const QMap<QString, QVariant> &data, Sink::Storage::Transaction &)> createEntity)
97{
98 auto time = QSharedPointer<QTime>::create();
99 time->start();
100 //TODO find items to remove
101 int count = 0;
102 for (auto it = data.constBegin(); it != data.constEnd(); it++) {
103 count++;
104 const auto remoteId = it.key().toUtf8();
105 auto entity = createEntity(remoteId, it.value(), synchronizationTransaction);
106 createOrModify(transaction, synchronizationTransaction, adaptorFactory, bufferType, remoteId, *entity);
107 }
108 Trace() << "Sync of " << count << " entities of type " << bufferType << " done." << Sink::Log::TraceTime(time->elapsed());
109}
110
111KAsync::Job<void> DummyResource::synchronizeWithSource(Sink::Storage &mainStore, Sink::Storage &synchronizationStore)
112{
113 Log() << " Synchronizing";
114 return KAsync::start<void>([this, &mainStore, &synchronizationStore]() {
115 auto transaction = mainStore.createTransaction(Sink::Storage::ReadOnly);
116 auto synchronizationTransaction = synchronizationStore.createTransaction(Sink::Storage::ReadWrite);
117 synchronize(ENTITY_TYPE_EVENT, DummyStore::instance().events(), transaction, synchronizationTransaction, *mEventAdaptorFactory, [this](const QByteArray &ridBuffer, const QMap<QString, QVariant> &data, Sink::Storage::Transaction &synchronizationTransaction) {
118 return createEvent(ridBuffer, data, synchronizationTransaction);
119 });
120 synchronize(ENTITY_TYPE_MAIL, DummyStore::instance().mails(), transaction, synchronizationTransaction, *mMailAdaptorFactory, [this](const QByteArray &ridBuffer, const QMap<QString, QVariant> &data, Sink::Storage::Transaction &synchronizationTransaction) {
121 return createMail(ridBuffer, data, synchronizationTransaction);
122 });
123 synchronize(ENTITY_TYPE_FOLDER, DummyStore::instance().folders(), transaction, synchronizationTransaction, *mFolderAdaptorFactory, [this](const QByteArray &ridBuffer, const QMap<QString, QVariant> &data, Sink::Storage::Transaction &synchronizationTransaction) {
124 return createFolder(ridBuffer, data, synchronizationTransaction);
125 });
126 Log() << "Done Synchronizing";
127 });
128}
129
130KAsync::Job<void> DummyResource::replay(Sink::Storage &synchronizationStore, const QByteArray &type, const QByteArray &key, const QByteArray &value)
131{
132 Trace() << "Replaying " << key;
133 return KAsync::null<void>();
134} 143}
135 144
136void DummyResource::removeDataFromDisk() 145void DummyResource::removeDataFromDisk()
@@ -160,6 +169,7 @@ KAsync::Job<void> DummyResource::inspect(int inspectionType, const QByteArray &i
160 return KAsync::null<void>(); 169 return KAsync::null<void>();
161} 170}
162 171
172
163DummyResourceFactory::DummyResourceFactory(QObject *parent) 173DummyResourceFactory::DummyResourceFactory(QObject *parent)
164 : Sink::ResourceFactory(parent) 174 : Sink::ResourceFactory(parent)
165{ 175{
@@ -178,3 +188,10 @@ void DummyResourceFactory::registerFacades(Sink::FacadeFactory &factory)
178 factory.registerFacade<Sink::ApplicationDomain::Folder, DummyResourceFolderFacade>(PLUGIN_NAME); 188 factory.registerFacade<Sink::ApplicationDomain::Folder, DummyResourceFolderFacade>(PLUGIN_NAME);
179} 189}
180 190
191void DummyResourceFactory::registerAdaptorFactories(Sink::AdaptorFactoryRegistry &registry)
192{
193 registry.registerFactory<Sink::ApplicationDomain::Folder, DummyFolderAdaptorFactory>(PLUGIN_NAME);
194 registry.registerFactory<Sink::ApplicationDomain::Mail, DummyMailAdaptorFactory>(PLUGIN_NAME);
195 registry.registerFactory<Sink::ApplicationDomain::Event, DummyEventAdaptorFactory>(PLUGIN_NAME);
196}
197
diff --git a/examples/dummyresource/resourcefactory.h b/examples/dummyresource/resourcefactory.h
index 865f6e5..f73eb32 100644
--- a/examples/dummyresource/resourcefactory.h
+++ b/examples/dummyresource/resourcefactory.h
@@ -37,13 +37,12 @@ class DummyResource : public Sink::GenericResource
37{ 37{
38public: 38public:
39 DummyResource(const QByteArray &instanceIdentifier, const QSharedPointer<Sink::Pipeline> &pipeline = QSharedPointer<Sink::Pipeline>()); 39 DummyResource(const QByteArray &instanceIdentifier, const QSharedPointer<Sink::Pipeline> &pipeline = QSharedPointer<Sink::Pipeline>());
40 KAsync::Job<void> synchronizeWithSource(Sink::Storage &mainStore, Sink::Storage &synchronizationStore) Q_DECL_OVERRIDE; 40 virtual ~DummyResource();
41 using GenericResource::synchronizeWithSource; 41
42 void removeDataFromDisk() Q_DECL_OVERRIDE; 42 void removeDataFromDisk() Q_DECL_OVERRIDE;
43 static void removeFromDisk(const QByteArray &instanceIdentifier); 43 static void removeFromDisk(const QByteArray &instanceIdentifier);
44 KAsync::Job<void> inspect(int inspectionType, const QByteArray &inspectionId, const QByteArray &domainType, const QByteArray &entityId, const QByteArray &property, const QVariant &expectedValue) Q_DECL_OVERRIDE; 44 KAsync::Job<void> inspect(int inspectionType, const QByteArray &inspectionId, const QByteArray &domainType, const QByteArray &entityId, const QByteArray &property, const QVariant &expectedValue) Q_DECL_OVERRIDE;
45private: 45private:
46 KAsync::Job<void> replay(Sink::Storage &synchronizationStore, const QByteArray &type, const QByteArray &key, const QByteArray &value) Q_DECL_OVERRIDE;
47 Sink::ApplicationDomain::Event::Ptr createEvent(const QByteArray &rid, const QMap<QString, QVariant> &data, Sink::Storage::Transaction &); 46 Sink::ApplicationDomain::Event::Ptr createEvent(const QByteArray &rid, const QMap<QString, QVariant> &data, Sink::Storage::Transaction &);
48 Sink::ApplicationDomain::Mail::Ptr createMail(const QByteArray &rid, const QMap<QString, QVariant> &data, Sink::Storage::Transaction &); 47 Sink::ApplicationDomain::Mail::Ptr createMail(const QByteArray &rid, const QMap<QString, QVariant> &data, Sink::Storage::Transaction &);
49 Sink::ApplicationDomain::Folder::Ptr createFolder(const QByteArray &rid, const QMap<QString, QVariant> &data, Sink::Storage::Transaction &); 48 Sink::ApplicationDomain::Folder::Ptr createFolder(const QByteArray &rid, const QMap<QString, QVariant> &data, Sink::Storage::Transaction &);
@@ -65,5 +64,6 @@ public:
65 64
66 Sink::Resource *createResource(const QByteArray &instanceIdentifier) Q_DECL_OVERRIDE; 65 Sink::Resource *createResource(const QByteArray &instanceIdentifier) Q_DECL_OVERRIDE;
67 void registerFacades(Sink::FacadeFactory &factory) Q_DECL_OVERRIDE; 66 void registerFacades(Sink::FacadeFactory &factory) Q_DECL_OVERRIDE;
67 void registerAdaptorFactories(Sink::AdaptorFactoryRegistry &registry) Q_DECL_OVERRIDE;
68}; 68};
69 69
diff --git a/examples/maildirresource/maildirresource.h b/examples/maildirresource/maildirresource.h
index e58dc16..30c565c 100644
--- a/examples/maildirresource/maildirresource.h
+++ b/examples/maildirresource/maildirresource.h
@@ -74,5 +74,6 @@ public:
74 74
75 Sink::Resource *createResource(const QByteArray &instanceIdentifier) Q_DECL_OVERRIDE; 75 Sink::Resource *createResource(const QByteArray &instanceIdentifier) Q_DECL_OVERRIDE;
76 void registerFacades(Sink::FacadeFactory &factory) Q_DECL_OVERRIDE; 76 void registerFacades(Sink::FacadeFactory &factory) Q_DECL_OVERRIDE;
77 void registerDomainTypeAdaptors(Sink::AdaptorFactory &factory) Q_DECL_OVERRIDE;
77}; 78};
78 79
diff --git a/tests/dummyresourcetest.cpp b/tests/dummyresourcetest.cpp
index 58df2da..d41f235 100644
--- a/tests/dummyresourcetest.cpp
+++ b/tests/dummyresourcetest.cpp
@@ -11,6 +11,7 @@
11#include "modelresult.h" 11#include "modelresult.h"
12#include "pipeline.h" 12#include "pipeline.h"
13#include "log.h" 13#include "log.h"
14#include "test.h"
14 15
15using namespace Sink; 16using namespace Sink;
16 17
@@ -28,6 +29,7 @@ class DummyResourceTest : public QObject
28private slots: 29private slots:
29 void initTestCase() 30 void initTestCase()
30 { 31 {
32 Sink::Test::initTest();
31 Sink::Log::setDebugOutputLevel(Sink::Log::Trace); 33 Sink::Log::setDebugOutputLevel(Sink::Log::Trace);
32 auto factory = Sink::ResourceFactory::load("org.kde.dummy"); 34 auto factory = Sink::ResourceFactory::load("org.kde.dummy");
33 QVERIFY(factory); 35 QVERIFY(factory);
diff --git a/tests/mailquerybenchmark.cpp b/tests/mailquerybenchmark.cpp
index 20ee63c..4e58899 100644
--- a/tests/mailquerybenchmark.cpp
+++ b/tests/mailquerybenchmark.cpp
@@ -33,6 +33,7 @@
33#include <common/pipeline.h> 33#include <common/pipeline.h>
34#include <common/index.h> 34#include <common/index.h>
35#include <common/indexupdater.h> 35#include <common/indexupdater.h>
36#include <common/adaptorfactoryregistry.h>
36 37
37#include "hawd/dataset.h" 38#include "hawd/dataset.h"
38#include "hawd/formatter.h" 39#include "hawd/formatter.h"
@@ -59,12 +60,11 @@ class MailQueryBenchmark : public QObject
59 TestResource::removeFromDisk(resourceIdentifier); 60 TestResource::removeFromDisk(resourceIdentifier);
60 61
61 auto pipeline = QSharedPointer<Sink::Pipeline>::create(resourceIdentifier); 62 auto pipeline = QSharedPointer<Sink::Pipeline>::create(resourceIdentifier);
63 pipeline->setResourceType("test");
62 64
63 auto mailFactory = QSharedPointer<TestMailAdaptorFactory>::create();
64 auto indexer = QSharedPointer<DefaultIndexUpdater<Sink::ApplicationDomain::Mail>>::create(); 65 auto indexer = QSharedPointer<DefaultIndexUpdater<Sink::ApplicationDomain::Mail>>::create();
65 66
66 pipeline->setPreprocessors("mail", QVector<Sink::Preprocessor *>() << indexer.data()); 67 pipeline->setPreprocessors("mail", QVector<Sink::Preprocessor *>() << indexer.data());
67 pipeline->setAdaptorFactory("mail", mailFactory);
68 68
69 auto domainTypeAdaptorFactory = QSharedPointer<TestMailAdaptorFactory>::create(); 69 auto domainTypeAdaptorFactory = QSharedPointer<TestMailAdaptorFactory>::create();
70 70
@@ -149,6 +149,7 @@ private slots:
149 void init() 149 void init()
150 { 150 {
151 resourceIdentifier = "org.kde.test.instance1"; 151 resourceIdentifier = "org.kde.test.instance1";
152 Sink::AdaptorFactoryRegistry::instance().registerFactory<Sink::ApplicationDomain::Mail, TestMailAdaptorFactory>("test");
152 } 153 }
153 154
154 void test50k() 155 void test50k()
diff --git a/tests/pipelinebenchmark.cpp b/tests/pipelinebenchmark.cpp
index 0133a6c..51481fd 100644
--- a/tests/pipelinebenchmark.cpp
+++ b/tests/pipelinebenchmark.cpp
@@ -33,6 +33,7 @@
33#include <common/pipeline.h> 33#include <common/pipeline.h>
34#include <common/index.h> 34#include <common/index.h>
35#include <common/indexupdater.h> 35#include <common/indexupdater.h>
36#include <common/adaptorfactoryregistry.h>
36 37
37#include "hawd/dataset.h" 38#include "hawd/dataset.h"
38#include "hawd/formatter.h" 39#include "hawd/formatter.h"
@@ -83,10 +84,8 @@ class PipelineBenchmark : public QObject
83 TestResource::removeFromDisk(resourceIdentifier); 84 TestResource::removeFromDisk(resourceIdentifier);
84 85
85 auto pipeline = QSharedPointer<Sink::Pipeline>::create(resourceIdentifier); 86 auto pipeline = QSharedPointer<Sink::Pipeline>::create(resourceIdentifier);
86
87 auto mailFactory = QSharedPointer<TestMailAdaptorFactory>::create();
88 pipeline->setPreprocessors("mail", preprocessors); 87 pipeline->setPreprocessors("mail", preprocessors);
89 pipeline->setAdaptorFactory("mail", mailFactory); 88 pipeline->setResourceType("test");
90 89
91 QTime time; 90 QTime time;
92 time.start(); 91 time.start();
@@ -131,6 +130,7 @@ private slots:
131 void init() 130 void init()
132 { 131 {
133 Sink::Log::setDebugOutputLevel(Sink::Log::Warning); 132 Sink::Log::setDebugOutputLevel(Sink::Log::Warning);
133 Sink::AdaptorFactoryRegistry::instance().registerFactory<Sink::ApplicationDomain::Mail, TestMailAdaptorFactory>("test");
134 resourceIdentifier = "org.kde.test.instance1"; 134 resourceIdentifier = "org.kde.test.instance1";
135 } 135 }
136 136
diff --git a/tests/pipelinetest.cpp b/tests/pipelinetest.cpp
index 9290050..6ea2041 100644
--- a/tests/pipelinetest.cpp
+++ b/tests/pipelinetest.cpp
@@ -19,6 +19,7 @@
19#include "log.h" 19#include "log.h"
20#include "domainadaptor.h" 20#include "domainadaptor.h"
21#include "definitions.h" 21#include "definitions.h"
22#include "adaptorfactoryregistry.h"
22 23
23static void removeFromDisk(const QString &name) 24static void removeFromDisk(const QString &name)
24{ 25{
@@ -185,6 +186,7 @@ private slots:
185 void initTestCase() 186 void initTestCase()
186 { 187 {
187 Sink::Log::setDebugOutputLevel(Sink::Log::Trace); 188 Sink::Log::setDebugOutputLevel(Sink::Log::Trace);
189 Sink::AdaptorFactoryRegistry::instance().registerFactory<Sink::ApplicationDomain::Event, TestEventAdaptorFactory>("test");
188 } 190 }
189 191
190 void init() 192 void init()
@@ -198,9 +200,7 @@ private slots:
198 auto command = createEntityCommand(createEvent(entityFbb)); 200 auto command = createEntityCommand(createEvent(entityFbb));
199 201
200 Sink::Pipeline pipeline("org.kde.pipelinetest.instance1"); 202 Sink::Pipeline pipeline("org.kde.pipelinetest.instance1");
201 203 pipeline.setResourceType("test");
202 auto adaptorFactory = QSharedPointer<TestEventAdaptorFactory>::create();
203 pipeline.setAdaptorFactory("event", adaptorFactory);
204 204
205 pipeline.startTransaction(); 205 pipeline.startTransaction();
206 pipeline.newEntity(command.constData(), command.size()); 206 pipeline.newEntity(command.constData(), command.size());
@@ -216,9 +216,9 @@ private slots:
216 auto command = createEntityCommand(createEvent(entityFbb, "summary", "description")); 216 auto command = createEntityCommand(createEvent(entityFbb, "summary", "description"));
217 217
218 Sink::Pipeline pipeline("org.kde.pipelinetest.instance1"); 218 Sink::Pipeline pipeline("org.kde.pipelinetest.instance1");
219 pipeline.setResourceType("test");
219 220
220 auto adaptorFactory = QSharedPointer<TestEventAdaptorFactory>::create(); 221 auto adaptorFactory = QSharedPointer<TestEventAdaptorFactory>::create();
221 pipeline.setAdaptorFactory("event", adaptorFactory);
222 222
223 // Create the initial revision 223 // Create the initial revision
224 pipeline.startTransaction(); 224 pipeline.startTransaction();
@@ -265,9 +265,9 @@ private slots:
265 auto command = createEntityCommand(createEvent(entityFbb)); 265 auto command = createEntityCommand(createEvent(entityFbb));
266 266
267 Sink::Pipeline pipeline("org.kde.pipelinetest.instance1"); 267 Sink::Pipeline pipeline("org.kde.pipelinetest.instance1");
268 pipeline.setResourceType("test");
268 269
269 auto adaptorFactory = QSharedPointer<TestEventAdaptorFactory>::create(); 270 auto adaptorFactory = QSharedPointer<TestEventAdaptorFactory>::create();
270 pipeline.setAdaptorFactory("event", adaptorFactory);
271 271
272 // Create the initial revision 272 // Create the initial revision
273 pipeline.startTransaction(); 273 pipeline.startTransaction();
@@ -309,7 +309,7 @@ private slots:
309 flatbuffers::FlatBufferBuilder entityFbb; 309 flatbuffers::FlatBufferBuilder entityFbb;
310 auto command = createEntityCommand(createEvent(entityFbb)); 310 auto command = createEntityCommand(createEvent(entityFbb));
311 Sink::Pipeline pipeline("org.kde.pipelinetest.instance1"); 311 Sink::Pipeline pipeline("org.kde.pipelinetest.instance1");
312 pipeline.setAdaptorFactory("event", QSharedPointer<TestEventAdaptorFactory>::create()); 312 pipeline.setResourceType("test");
313 313
314 // Create the initial revision 314 // Create the initial revision
315 pipeline.startTransaction(); 315 pipeline.startTransaction();
@@ -346,9 +346,9 @@ private slots:
346 TestProcessor testProcessor; 346 TestProcessor testProcessor;
347 347
348 Sink::Pipeline pipeline("org.kde.pipelinetest.instance1"); 348 Sink::Pipeline pipeline("org.kde.pipelinetest.instance1");
349 pipeline.setResourceType("test");
349 pipeline.setPreprocessors("event", QVector<Sink::Preprocessor *>() << &testProcessor); 350 pipeline.setPreprocessors("event", QVector<Sink::Preprocessor *>() << &testProcessor);
350 pipeline.startTransaction(); 351 pipeline.startTransaction();
351 pipeline.setAdaptorFactory("event", QSharedPointer<TestEventAdaptorFactory>::create());
352 352
353 // Actual test 353 // Actual test
354 { 354 {
diff --git a/tests/querytest.cpp b/tests/querytest.cpp
index a654931..6d7746e 100644
--- a/tests/querytest.cpp
+++ b/tests/querytest.cpp
@@ -9,6 +9,7 @@
9#include "resourceconfig.h" 9#include "resourceconfig.h"
10#include "log.h" 10#include "log.h"
11#include "modelresult.h" 11#include "modelresult.h"
12#include "test.h"
12 13
13/** 14/**
14 * Test of the query system using the dummy resource. 15 * Test of the query system using the dummy resource.
@@ -21,6 +22,7 @@ class QueryTest : public QObject
21private slots: 22private slots:
22 void initTestCase() 23 void initTestCase()
23 { 24 {
25 Sink::Test::initTest();
24 Sink::Log::setDebugOutputLevel(Sink::Log::Trace); 26 Sink::Log::setDebugOutputLevel(Sink::Log::Trace);
25 auto factory = Sink::ResourceFactory::load("org.kde.dummy"); 27 auto factory = Sink::ResourceFactory::load("org.kde.dummy");
26 QVERIFY(factory); 28 QVERIFY(factory);
diff --git a/tests/testimplementations.h b/tests/testimplementations.h
index 688875d..197602c 100644
--- a/tests/testimplementations.h
+++ b/tests/testimplementations.h
@@ -107,7 +107,7 @@ public:
107class TestResource : public Sink::GenericResource 107class TestResource : public Sink::GenericResource
108{ 108{
109public: 109public:
110 TestResource(const QByteArray &instanceIdentifier, QSharedPointer<Sink::Pipeline> pipeline) : Sink::GenericResource(instanceIdentifier, pipeline) 110 TestResource(const QByteArray &instanceIdentifier, QSharedPointer<Sink::Pipeline> pipeline) : Sink::GenericResource("test", instanceIdentifier, pipeline, QSharedPointer<Sink::ChangeReplay>(), QSharedPointer<Sink::Synchronizer>())
111 { 111 {
112 } 112 }
113 113
generated by cgit v1.2.3 (git 2.25.1) at 2025-04-29 05:21:21 +0000