diff options
author | Christian Mollekopf <chrigi_1@fastmail.fm> | 2016-11-21 23:13:38 +0100 |
---|---|---|
committer | Christian Mollekopf <chrigi_1@fastmail.fm> | 2016-11-21 23:51:24 +0100 |
commit | 1d713d9e2dbaf27de9da087f9270d260dfc40c31 (patch) | |
tree | 666d8edd42e44df3eaa674a35b6e938b99c2f4b4 | |
parent | 0adba61a00491b96dadaa6d4719cb46831356222 (diff) | |
download | sink-1d713d9e2dbaf27de9da087f9270d260dfc40c31.tar.gz sink-1d713d9e2dbaf27de9da087f9270d260dfc40c31.zip |
Folded the SourceWriteback into the Synchronizer.
By concentrating all communication to the source in one place we get rid
of several oddities.
* Quite a bit of duplication since both need access to the
synchronizationStore and the source.
* We currently have an akward locking in place because both classes
access the ync store. This is not easier to resolve cleanly.
* The live of resource implementers becomes easier.
* An implementation could elect to not use changereplay and always do a
full sync... (maybe?)
-rw-r--r-- | common/CMakeLists.txt | 1 | ||||
-rw-r--r-- | common/genericresource.cpp | 29 | ||||
-rw-r--r-- | common/genericresource.h | 4 | ||||
-rw-r--r-- | common/sourcewriteback.cpp | 146 | ||||
-rw-r--r-- | common/sourcewriteback.h | 71 | ||||
-rw-r--r-- | common/synchronizer.cpp | 94 | ||||
-rw-r--r-- | common/synchronizer.h | 15 | ||||
-rw-r--r-- | examples/dummyresource/resourcefactory.cpp | 3 | ||||
-rw-r--r-- | examples/imapresource/imapresource.cpp | 23 | ||||
-rw-r--r-- | examples/maildirresource/maildirresource.cpp | 16 | ||||
-rw-r--r-- | examples/mailtransportresource/mailtransportresource.cpp | 40 |
11 files changed, 133 insertions, 309 deletions
diff --git a/common/CMakeLists.txt b/common/CMakeLists.txt index a08be8a..5ba524b 100644 --- a/common/CMakeLists.txt +++ b/common/CMakeLists.txt | |||
@@ -70,7 +70,6 @@ set(command_SRCS | |||
70 | adaptorfactoryregistry.cpp | 70 | adaptorfactoryregistry.cpp |
71 | synchronizer.cpp | 71 | synchronizer.cpp |
72 | remoteidmap.cpp | 72 | remoteidmap.cpp |
73 | sourcewriteback.cpp | ||
74 | mailpreprocessor.cpp | 73 | mailpreprocessor.cpp |
75 | specialpurposepreprocessor.cpp | 74 | specialpurposepreprocessor.cpp |
76 | datastorequery.cpp | 75 | datastorequery.cpp |
diff --git a/common/genericresource.cpp b/common/genericresource.cpp index c4c8bc6..746fa33 100644 --- a/common/genericresource.cpp +++ b/common/genericresource.cpp | |||
@@ -292,14 +292,14 @@ KAsync::Job<void> GenericResource::inspect( | |||
292 | 292 | ||
293 | void GenericResource::enableChangeReplay(bool enable) | 293 | void GenericResource::enableChangeReplay(bool enable) |
294 | { | 294 | { |
295 | Q_ASSERT(mChangeReplay); | 295 | Q_ASSERT(mSynchronizer); |
296 | if (enable) { | 296 | if (enable) { |
297 | QObject::connect(mPipeline.data(), &Pipeline::revisionUpdated, mChangeReplay.data(), &ChangeReplay::revisionChanged, Qt::QueuedConnection); | 297 | QObject::connect(mPipeline.data(), &Pipeline::revisionUpdated, mSynchronizer.data(), &ChangeReplay::revisionChanged, Qt::QueuedConnection); |
298 | QObject::connect(mChangeReplay.data(), &ChangeReplay::changesReplayed, this, &GenericResource::updateLowerBoundRevision); | 298 | QObject::connect(mSynchronizer.data(), &ChangeReplay::changesReplayed, this, &GenericResource::updateLowerBoundRevision); |
299 | QMetaObject::invokeMethod(mChangeReplay.data(), "revisionChanged", Qt::QueuedConnection); | 299 | QMetaObject::invokeMethod(mSynchronizer.data(), "revisionChanged", Qt::QueuedConnection); |
300 | } else { | 300 | } else { |
301 | QObject::disconnect(mPipeline.data(), &Pipeline::revisionUpdated, mChangeReplay.data(), &ChangeReplay::revisionChanged); | 301 | QObject::disconnect(mPipeline.data(), &Pipeline::revisionUpdated, mSynchronizer.data(), &ChangeReplay::revisionChanged); |
302 | QObject::disconnect(mChangeReplay.data(), &ChangeReplay::changesReplayed, this, &GenericResource::updateLowerBoundRevision); | 302 | QObject::disconnect(mSynchronizer.data(), &ChangeReplay::changesReplayed, this, &GenericResource::updateLowerBoundRevision); |
303 | } | 303 | } |
304 | } | 304 | } |
305 | 305 | ||
@@ -314,13 +314,8 @@ void GenericResource::setupSynchronizer(const QSharedPointer<Synchronizer> &sync | |||
314 | mSynchronizer->setup([this](int commandId, const QByteArray &data) { | 314 | mSynchronizer->setup([this](int commandId, const QByteArray &data) { |
315 | enqueueCommand(mSynchronizerQueue, commandId, data); | 315 | enqueueCommand(mSynchronizerQueue, commandId, data); |
316 | }, mSynchronizerQueue); | 316 | }, mSynchronizerQueue); |
317 | } | ||
318 | |||
319 | void GenericResource::setupChangereplay(const QSharedPointer<ChangeReplay> &changeReplay) | ||
320 | { | ||
321 | mChangeReplay = changeReplay; | ||
322 | { | 317 | { |
323 | auto ret = QObject::connect(mChangeReplay.data(), &ChangeReplay::replayingChanges, [this]() { | 318 | auto ret = QObject::connect(mSynchronizer.data(), &Synchronizer::replayingChanges, [this]() { |
324 | Sink::Notification n; | 319 | Sink::Notification n; |
325 | n.id = "changereplay"; | 320 | n.id = "changereplay"; |
326 | n.type = Sink::Notification::Status; | 321 | n.type = Sink::Notification::Status; |
@@ -331,7 +326,7 @@ void GenericResource::setupChangereplay(const QSharedPointer<ChangeReplay> &chan | |||
331 | Q_ASSERT(ret); | 326 | Q_ASSERT(ret); |
332 | } | 327 | } |
333 | { | 328 | { |
334 | auto ret = QObject::connect(mChangeReplay.data(), &ChangeReplay::changesReplayed, [this]() { | 329 | auto ret = QObject::connect(mSynchronizer.data(), &Synchronizer::changesReplayed, [this]() { |
335 | Sink::Notification n; | 330 | Sink::Notification n; |
336 | n.id = "changereplay"; | 331 | n.id = "changereplay"; |
337 | n.type = Sink::Notification::Status; | 332 | n.type = Sink::Notification::Status; |
@@ -342,7 +337,7 @@ void GenericResource::setupChangereplay(const QSharedPointer<ChangeReplay> &chan | |||
342 | Q_ASSERT(ret); | 337 | Q_ASSERT(ret); |
343 | } | 338 | } |
344 | 339 | ||
345 | mProcessor->setOldestUsedRevision(mChangeReplay->getLastReplayedRevision()); | 340 | mProcessor->setOldestUsedRevision(mSynchronizer->getLastReplayedRevision()); |
346 | enableChangeReplay(true); | 341 | enableChangeReplay(true); |
347 | } | 342 | } |
348 | 343 | ||
@@ -459,11 +454,11 @@ KAsync::Job<void> GenericResource::processAllMessages() | |||
459 | .then<void>([this](KAsync::Future<void> &f) { waitForDrained(f, mSynchronizerQueue); }) | 454 | .then<void>([this](KAsync::Future<void> &f) { waitForDrained(f, mSynchronizerQueue); }) |
460 | .then<void>([this](KAsync::Future<void> &f) { waitForDrained(f, mUserQueue); }) | 455 | .then<void>([this](KAsync::Future<void> &f) { waitForDrained(f, mUserQueue); }) |
461 | .then<void>([this](KAsync::Future<void> &f) { | 456 | .then<void>([this](KAsync::Future<void> &f) { |
462 | if (mChangeReplay->allChangesReplayed()) { | 457 | if (mSynchronizer->allChangesReplayed()) { |
463 | f.setFinished(); | 458 | f.setFinished(); |
464 | } else { | 459 | } else { |
465 | auto context = new QObject; | 460 | auto context = new QObject; |
466 | QObject::connect(mChangeReplay.data(), &ChangeReplay::changesReplayed, context, [&f, context]() { | 461 | QObject::connect(mSynchronizer.data(), &ChangeReplay::changesReplayed, context, [&f, context]() { |
467 | delete context; | 462 | delete context; |
468 | f.setFinished(); | 463 | f.setFinished(); |
469 | }); | 464 | }); |
@@ -473,7 +468,7 @@ KAsync::Job<void> GenericResource::processAllMessages() | |||
473 | 468 | ||
474 | void GenericResource::updateLowerBoundRevision() | 469 | void GenericResource::updateLowerBoundRevision() |
475 | { | 470 | { |
476 | mProcessor->setOldestUsedRevision(qMin(mClientLowerBoundRevision, mChangeReplay->getLastReplayedRevision())); | 471 | mProcessor->setOldestUsedRevision(qMin(mClientLowerBoundRevision, mSynchronizer->getLastReplayedRevision())); |
477 | } | 472 | } |
478 | 473 | ||
479 | void GenericResource::setLowerBoundRevision(qint64 revision) | 474 | void GenericResource::setLowerBoundRevision(qint64 revision) |
diff --git a/common/genericresource.h b/common/genericresource.h index 3736c8f..7e0f5ad 100644 --- a/common/genericresource.h +++ b/common/genericresource.h | |||
@@ -24,7 +24,7 @@ | |||
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" | 27 | #include <resourcecontext.h> |
28 | 28 | ||
29 | #include <QTimer> | 29 | #include <QTimer> |
30 | 30 | ||
@@ -66,7 +66,6 @@ protected: | |||
66 | 66 | ||
67 | void setupPreprocessors(const QByteArray &type, const QVector<Sink::Preprocessor *> &preprocessors); | 67 | void setupPreprocessors(const QByteArray &type, const QVector<Sink::Preprocessor *> &preprocessors); |
68 | void setupSynchronizer(const QSharedPointer<Synchronizer> &synchronizer); | 68 | void setupSynchronizer(const QSharedPointer<Synchronizer> &synchronizer); |
69 | void setupChangereplay(const QSharedPointer<ChangeReplay> &changeReplay); | ||
70 | 69 | ||
71 | void onProcessorError(int errorCode, const QString &errorMessage); | 70 | void onProcessorError(int errorCode, const QString &errorMessage); |
72 | void enqueueCommand(MessageQueue &mq, int commandId, const QByteArray &data); | 71 | void enqueueCommand(MessageQueue &mq, int commandId, const QByteArray &data); |
@@ -78,7 +77,6 @@ protected: | |||
78 | 77 | ||
79 | private: | 78 | private: |
80 | std::unique_ptr<CommandProcessor> mProcessor; | 79 | std::unique_ptr<CommandProcessor> mProcessor; |
81 | QSharedPointer<ChangeReplay> mChangeReplay; | ||
82 | QSharedPointer<Synchronizer> mSynchronizer; | 80 | QSharedPointer<Synchronizer> mSynchronizer; |
83 | int mError; | 81 | int mError; |
84 | QTimer mCommitQueueTimer; | 82 | QTimer mCommitQueueTimer; |
diff --git a/common/sourcewriteback.cpp b/common/sourcewriteback.cpp deleted file mode 100644 index e2994d1..0000000 --- a/common/sourcewriteback.cpp +++ /dev/null | |||
@@ -1,146 +0,0 @@ | |||
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 "sourcewriteback.h" | ||
21 | |||
22 | #include "definitions.h" | ||
23 | #include "log.h" | ||
24 | #include "bufferutils.h" | ||
25 | #include "entitybuffer.h" | ||
26 | #include "entity_generated.h" | ||
27 | |||
28 | #define ENTITY_TYPE_MAIL "mail" | ||
29 | #define ENTITY_TYPE_FOLDER "folder" | ||
30 | |||
31 | SINK_DEBUG_AREA("sourcewriteback") | ||
32 | |||
33 | using namespace Sink; | ||
34 | |||
35 | SourceWriteBack::SourceWriteBack(const ResourceContext &context) | ||
36 | : ChangeReplay(context), | ||
37 | mResourceContext(context), | ||
38 | mSyncStorage(Sink::storageLocation(), context.instanceId() + ".synchronization", Sink::Storage::DataStore::ReadWrite), | ||
39 | mEntityStore(QSharedPointer<Storage::EntityStore>::create(mResourceContext)) | ||
40 | { | ||
41 | |||
42 | } | ||
43 | |||
44 | Storage::EntityStore &SourceWriteBack::store() | ||
45 | { | ||
46 | return *mEntityStore; | ||
47 | } | ||
48 | |||
49 | RemoteIdMap &SourceWriteBack::syncStore() | ||
50 | { | ||
51 | if (!mSyncStore) { | ||
52 | mSyncStore = QSharedPointer<RemoteIdMap>::create(mSyncTransaction); | ||
53 | } | ||
54 | return *mSyncStore; | ||
55 | } | ||
56 | |||
57 | bool SourceWriteBack::canReplay(const QByteArray &type, const QByteArray &key, const QByteArray &value) | ||
58 | { | ||
59 | Sink::EntityBuffer buffer(value); | ||
60 | const Sink::Entity &entity = buffer.entity(); | ||
61 | const auto metadataBuffer = Sink::EntityBuffer::readBuffer<Sink::Metadata>(entity.metadata()); | ||
62 | Q_ASSERT(metadataBuffer); | ||
63 | if (!metadataBuffer->replayToSource()) { | ||
64 | SinkTrace() << "Change is coming from the source"; | ||
65 | } | ||
66 | return metadataBuffer->replayToSource(); | ||
67 | } | ||
68 | |||
69 | KAsync::Job<void> SourceWriteBack::replay(const QByteArray &type, const QByteArray &key, const QByteArray &value) | ||
70 | { | ||
71 | SinkTrace() << "Replaying" << type << key; | ||
72 | |||
73 | Sink::EntityBuffer buffer(value); | ||
74 | const Sink::Entity &entity = buffer.entity(); | ||
75 | const auto metadataBuffer = Sink::EntityBuffer::readBuffer<Sink::Metadata>(entity.metadata()); | ||
76 | Q_ASSERT(metadataBuffer); | ||
77 | Q_ASSERT(!mSyncStore); | ||
78 | Q_ASSERT(!mSyncTransaction); | ||
79 | mEntityStore->startTransaction(Storage::DataStore::ReadOnly); | ||
80 | mSyncTransaction = mSyncStorage.createTransaction(Sink::Storage::DataStore::ReadWrite); | ||
81 | |||
82 | // const qint64 revision = metadataBuffer ? metadataBuffer->revision() : -1; | ||
83 | const auto operation = metadataBuffer ? metadataBuffer->operation() : Sink::Operation_Creation; | ||
84 | const auto uid = Sink::Storage::DataStore::uidFromKey(key); | ||
85 | const auto modifiedProperties = metadataBuffer->modifiedProperties() ? BufferUtils::fromVector(*metadataBuffer->modifiedProperties()) : QByteArrayList(); | ||
86 | QByteArray oldRemoteId; | ||
87 | |||
88 | if (operation != Sink::Operation_Creation) { | ||
89 | oldRemoteId = syncStore().resolveLocalId(type, uid); | ||
90 | if (oldRemoteId.isEmpty()) { | ||
91 | SinkWarning() << "Couldn't find the remote id for: " << type << uid; | ||
92 | return KAsync::error<void>(1, "Couldn't find the remote id."); | ||
93 | } | ||
94 | } | ||
95 | SinkTrace() << "Replaying " << key << type << uid << oldRemoteId; | ||
96 | |||
97 | KAsync::Job<QByteArray> job = KAsync::null<QByteArray>(); | ||
98 | if (type == ENTITY_TYPE_FOLDER) { | ||
99 | auto folder = store().readEntity<ApplicationDomain::Folder>(key); | ||
100 | job = replay(folder, operation, oldRemoteId, modifiedProperties); | ||
101 | } else if (type == ENTITY_TYPE_MAIL) { | ||
102 | auto mail = store().readEntity<ApplicationDomain::Mail>(key); | ||
103 | job = replay(mail, operation, oldRemoteId, modifiedProperties); | ||
104 | } | ||
105 | |||
106 | return job.syncThen<void, QByteArray>([this, operation, type, uid, oldRemoteId](const QByteArray &remoteId) { | ||
107 | if (operation == Sink::Operation_Creation) { | ||
108 | SinkTrace() << "Replayed creation with remote id: " << remoteId; | ||
109 | if (remoteId.isEmpty()) { | ||
110 | SinkWarning() << "Returned an empty remoteId from the creation"; | ||
111 | } else { | ||
112 | syncStore().recordRemoteId(type, uid, remoteId); | ||
113 | } | ||
114 | } else if (operation == Sink::Operation_Modification) { | ||
115 | SinkTrace() << "Replayed modification with remote id: " << remoteId; | ||
116 | if (remoteId.isEmpty()) { | ||
117 | SinkWarning() << "Returned an empty remoteId from the creation"; | ||
118 | } else { | ||
119 | syncStore().updateRemoteId(type, uid, remoteId); | ||
120 | } | ||
121 | } else if (operation == Sink::Operation_Removal) { | ||
122 | SinkTrace() << "Replayed removal with remote id: " << oldRemoteId; | ||
123 | syncStore().removeRemoteId(type, uid, oldRemoteId); | ||
124 | } else { | ||
125 | SinkError() << "Unkown operation" << operation; | ||
126 | } | ||
127 | }) | ||
128 | .syncThen<void>([this](const KAsync::Error &error) { | ||
129 | if (error) { | ||
130 | SinkWarning() << "Failed to replay change: " << error.errorMessage; | ||
131 | } | ||
132 | mSyncStore.clear(); | ||
133 | mSyncTransaction.commit(); | ||
134 | mEntityStore->abortTransaction(); | ||
135 | }); | ||
136 | } | ||
137 | |||
138 | KAsync::Job<QByteArray> SourceWriteBack::replay(const ApplicationDomain::Mail &, Sink::Operation, const QByteArray &, const QList<QByteArray> &) | ||
139 | { | ||
140 | return KAsync::null<QByteArray>(); | ||
141 | } | ||
142 | |||
143 | KAsync::Job<QByteArray> SourceWriteBack::replay(const ApplicationDomain::Folder &, Sink::Operation, const QByteArray &, const QList<QByteArray> &) | ||
144 | { | ||
145 | return KAsync::null<QByteArray>(); | ||
146 | } | ||
diff --git a/common/sourcewriteback.h b/common/sourcewriteback.h deleted file mode 100644 index cf393e4..0000000 --- a/common/sourcewriteback.h +++ /dev/null | |||
@@ -1,71 +0,0 @@ | |||
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 | |||
24 | #include "changereplay.h" | ||
25 | #include "storage.h" | ||
26 | #include "storage/entitystore.h" | ||
27 | #include "remoteidmap.h" | ||
28 | #include "metadata_generated.h" | ||
29 | |||
30 | namespace Sink { | ||
31 | |||
32 | /** | ||
33 | * Replay changes to the source | ||
34 | */ | ||
35 | class SINK_EXPORT SourceWriteBack : public ChangeReplay | ||
36 | { | ||
37 | public: | ||
38 | SourceWriteBack(const ResourceContext &resourceContext); | ||
39 | |||
40 | protected: | ||
41 | ///Base implementation calls the replay$Type calls | ||
42 | virtual KAsync::Job<void> replay(const QByteArray &type, const QByteArray &key, const QByteArray &value) Q_DECL_OVERRIDE; | ||
43 | virtual bool canReplay(const QByteArray &type, const QByteArray &key, const QByteArray &value) Q_DECL_OVERRIDE; | ||
44 | |||
45 | protected: | ||
46 | ///Implement to write back changes to the server | ||
47 | virtual KAsync::Job<QByteArray> replay(const Sink::ApplicationDomain::Mail &, Sink::Operation, const QByteArray &oldRemoteId, const QList<QByteArray> &); | ||
48 | virtual KAsync::Job<QByteArray> replay(const Sink::ApplicationDomain::Folder &, Sink::Operation, const QByteArray &oldRemoteId, const QList<QByteArray> &); | ||
49 | |||
50 | //Read/Write access to sync storage | ||
51 | RemoteIdMap &syncStore(); | ||
52 | |||
53 | template <typename T> | ||
54 | T getPrevious(const T &entity) | ||
55 | { | ||
56 | return store().readPrevious<T>(entity.identifier(), entity.revision()); | ||
57 | } | ||
58 | |||
59 | private: | ||
60 | //Read only access to main storage | ||
61 | Storage::EntityStore &store(); | ||
62 | ResourceContext mResourceContext; | ||
63 | Sink::Storage::DataStore mSyncStorage; | ||
64 | QSharedPointer<RemoteIdMap> mSyncStore; | ||
65 | QSharedPointer<Storage::EntityStore> mEntityStore; | ||
66 | Sink::Storage::DataStore::Transaction mSyncTransaction; | ||
67 | QByteArray mResourceType; | ||
68 | QByteArray mResourceInstanceIdentifier; | ||
69 | }; | ||
70 | |||
71 | } | ||
diff --git a/common/synchronizer.cpp b/common/synchronizer.cpp index 713387e..10acefc 100644 --- a/common/synchronizer.cpp +++ b/common/synchronizer.cpp | |||
@@ -33,7 +33,8 @@ SINK_DEBUG_AREA("synchronizer") | |||
33 | using namespace Sink; | 33 | using namespace Sink; |
34 | 34 | ||
35 | Synchronizer::Synchronizer(const Sink::ResourceContext &context) | 35 | Synchronizer::Synchronizer(const Sink::ResourceContext &context) |
36 | : mResourceContext(context), | 36 | : ChangeReplay(context), |
37 | mResourceContext(context), | ||
37 | mEntityStore(Storage::EntityStore::Ptr::create(mResourceContext)), | 38 | mEntityStore(Storage::EntityStore::Ptr::create(mResourceContext)), |
38 | mSyncStorage(Sink::storageLocation(), mResourceContext.instanceId() + ".synchronization", Sink::Storage::DataStore::DataStore::ReadWrite) | 39 | mSyncStorage(Sink::storageLocation(), mResourceContext.instanceId() + ".synchronization", Sink::Storage::DataStore::DataStore::ReadWrite) |
39 | { | 40 | { |
@@ -310,6 +311,97 @@ Sink::Storage::DataStore::DataStore::Transaction &Synchronizer::syncTransaction( | |||
310 | return mSyncTransaction; | 311 | return mSyncTransaction; |
311 | } | 312 | } |
312 | 313 | ||
314 | bool Synchronizer::canReplay(const QByteArray &type, const QByteArray &key, const QByteArray &value) | ||
315 | { | ||
316 | Sink::EntityBuffer buffer(value); | ||
317 | const Sink::Entity &entity = buffer.entity(); | ||
318 | const auto metadataBuffer = Sink::EntityBuffer::readBuffer<Sink::Metadata>(entity.metadata()); | ||
319 | Q_ASSERT(metadataBuffer); | ||
320 | if (!metadataBuffer->replayToSource()) { | ||
321 | SinkTrace() << "Change is coming from the source"; | ||
322 | } | ||
323 | return metadataBuffer->replayToSource(); | ||
324 | } | ||
325 | |||
326 | KAsync::Job<void> Synchronizer::replay(const QByteArray &type, const QByteArray &key, const QByteArray &value) | ||
327 | { | ||
328 | SinkTrace() << "Replaying" << type << key; | ||
329 | |||
330 | Sink::EntityBuffer buffer(value); | ||
331 | const Sink::Entity &entity = buffer.entity(); | ||
332 | const auto metadataBuffer = Sink::EntityBuffer::readBuffer<Sink::Metadata>(entity.metadata()); | ||
333 | Q_ASSERT(metadataBuffer); | ||
334 | Q_ASSERT(!mSyncStore); | ||
335 | Q_ASSERT(!mSyncTransaction); | ||
336 | mEntityStore->startTransaction(Storage::DataStore::ReadOnly); | ||
337 | mSyncTransaction = mSyncStorage.createTransaction(Sink::Storage::DataStore::ReadWrite); | ||
338 | |||
339 | const auto operation = metadataBuffer ? metadataBuffer->operation() : Sink::Operation_Creation; | ||
340 | const auto uid = Sink::Storage::DataStore::uidFromKey(key); | ||
341 | const auto modifiedProperties = metadataBuffer->modifiedProperties() ? BufferUtils::fromVector(*metadataBuffer->modifiedProperties()) : QByteArrayList(); | ||
342 | QByteArray oldRemoteId; | ||
343 | |||
344 | if (operation != Sink::Operation_Creation) { | ||
345 | oldRemoteId = syncStore().resolveLocalId(type, uid); | ||
346 | if (oldRemoteId.isEmpty()) { | ||
347 | SinkWarning() << "Couldn't find the remote id for: " << type << uid; | ||
348 | return KAsync::error<void>(1, "Couldn't find the remote id."); | ||
349 | } | ||
350 | } | ||
351 | SinkTrace() << "Replaying " << key << type << uid << oldRemoteId; | ||
352 | |||
353 | KAsync::Job<QByteArray> job = KAsync::null<QByteArray>(); | ||
354 | //TODO This requires supporting every domain type here as well. Can we solve this better so we can do the dispatch somewhere centrally? | ||
355 | if (type == ApplicationDomain::getTypeName<ApplicationDomain::Folder>()) { | ||
356 | auto folder = store().readEntity<ApplicationDomain::Folder>(key); | ||
357 | job = replay(folder, operation, oldRemoteId, modifiedProperties); | ||
358 | } else if (type == ApplicationDomain::getTypeName<ApplicationDomain::Mail>()) { | ||
359 | auto mail = store().readEntity<ApplicationDomain::Mail>(key); | ||
360 | job = replay(mail, operation, oldRemoteId, modifiedProperties); | ||
361 | } | ||
362 | |||
363 | return job.syncThen<void, QByteArray>([this, operation, type, uid, oldRemoteId](const QByteArray &remoteId) { | ||
364 | if (operation == Sink::Operation_Creation) { | ||
365 | SinkTrace() << "Replayed creation with remote id: " << remoteId; | ||
366 | if (remoteId.isEmpty()) { | ||
367 | SinkWarning() << "Returned an empty remoteId from the creation"; | ||
368 | } else { | ||
369 | syncStore().recordRemoteId(type, uid, remoteId); | ||
370 | } | ||
371 | } else if (operation == Sink::Operation_Modification) { | ||
372 | SinkTrace() << "Replayed modification with remote id: " << remoteId; | ||
373 | if (remoteId.isEmpty()) { | ||
374 | SinkWarning() << "Returned an empty remoteId from the creation"; | ||
375 | } else { | ||
376 | syncStore().updateRemoteId(type, uid, remoteId); | ||
377 | } | ||
378 | } else if (operation == Sink::Operation_Removal) { | ||
379 | SinkTrace() << "Replayed removal with remote id: " << oldRemoteId; | ||
380 | syncStore().removeRemoteId(type, uid, oldRemoteId); | ||
381 | } else { | ||
382 | SinkError() << "Unkown operation" << operation; | ||
383 | } | ||
384 | }) | ||
385 | .syncThen<void>([this](const KAsync::Error &error) { | ||
386 | if (error) { | ||
387 | SinkWarning() << "Failed to replay change: " << error.errorMessage; | ||
388 | } | ||
389 | mSyncStore.clear(); | ||
390 | mSyncTransaction.commit(); | ||
391 | mEntityStore->abortTransaction(); | ||
392 | }); | ||
393 | } | ||
394 | |||
395 | KAsync::Job<QByteArray> Synchronizer::replay(const ApplicationDomain::Mail &, Sink::Operation, const QByteArray &, const QList<QByteArray> &) | ||
396 | { | ||
397 | return KAsync::null<QByteArray>(); | ||
398 | } | ||
399 | |||
400 | KAsync::Job<QByteArray> Synchronizer::replay(const ApplicationDomain::Folder &, Sink::Operation, const QByteArray &, const QList<QByteArray> &) | ||
401 | { | ||
402 | return KAsync::null<QByteArray>(); | ||
403 | } | ||
404 | |||
313 | #define REGISTER_TYPE(T) \ | 405 | #define REGISTER_TYPE(T) \ |
314 | template void Synchronizer::createOrModify(const QByteArray &bufferType, const QByteArray &remoteId, const T &entity, const QHash<QByteArray, Sink::Query::Comparator> &mergeCriteria); \ | 406 | template void Synchronizer::createOrModify(const QByteArray &bufferType, const QByteArray &remoteId, const T &entity, const QHash<QByteArray, Sink::Query::Comparator> &mergeCriteria); \ |
315 | template void Synchronizer::modify(const T &entity); | 407 | template void Synchronizer::modify(const T &entity); |
diff --git a/common/synchronizer.h b/common/synchronizer.h index 47518ee..0a51f54 100644 --- a/common/synchronizer.h +++ b/common/synchronizer.h | |||
@@ -27,6 +27,8 @@ | |||
27 | #include <messagequeue.h> | 27 | #include <messagequeue.h> |
28 | #include <storage.h> | 28 | #include <storage.h> |
29 | #include <storage/entitystore.h> | 29 | #include <storage/entitystore.h> |
30 | #include "changereplay.h" | ||
31 | #include "remoteidmap.h" | ||
30 | 32 | ||
31 | namespace Sink { | 33 | namespace Sink { |
32 | class RemoteIdMap; | 34 | class RemoteIdMap; |
@@ -34,8 +36,9 @@ class RemoteIdMap; | |||
34 | /** | 36 | /** |
35 | * Synchronize and add what we don't already have to local queue | 37 | * Synchronize and add what we don't already have to local queue |
36 | */ | 38 | */ |
37 | class SINK_EXPORT Synchronizer | 39 | class SINK_EXPORT Synchronizer : public ChangeReplay |
38 | { | 40 | { |
41 | Q_OBJECT | ||
39 | public: | 42 | public: |
40 | Synchronizer(const Sink::ResourceContext &resourceContext); | 43 | Synchronizer(const Sink::ResourceContext &resourceContext); |
41 | virtual ~Synchronizer(); | 44 | virtual ~Synchronizer(); |
@@ -53,6 +56,16 @@ public: | |||
53 | Sink::Storage::DataStore::Transaction &syncTransaction(); | 56 | Sink::Storage::DataStore::Transaction &syncTransaction(); |
54 | 57 | ||
55 | protected: | 58 | protected: |
59 | ///Base implementation calls the replay$Type calls | ||
60 | virtual KAsync::Job<void> replay(const QByteArray &type, const QByteArray &key, const QByteArray &value) Q_DECL_OVERRIDE; | ||
61 | virtual bool canReplay(const QByteArray &type, const QByteArray &key, const QByteArray &value) Q_DECL_OVERRIDE; | ||
62 | |||
63 | protected: | ||
64 | ///Implement to write back changes to the server | ||
65 | virtual KAsync::Job<QByteArray> replay(const Sink::ApplicationDomain::Mail &, Sink::Operation, const QByteArray &oldRemoteId, const QList<QByteArray> &); | ||
66 | virtual KAsync::Job<QByteArray> replay(const Sink::ApplicationDomain::Folder &, Sink::Operation, const QByteArray &oldRemoteId, const QList<QByteArray> &); | ||
67 | |||
68 | protected: | ||
56 | ///Calls the callback to enqueue the command | 69 | ///Calls the callback to enqueue the command |
57 | void enqueueCommand(int commandId, const QByteArray &data); | 70 | void enqueueCommand(int commandId, const QByteArray &data); |
58 | 71 | ||
diff --git a/examples/dummyresource/resourcefactory.cpp b/examples/dummyresource/resourcefactory.cpp index 46e67f3..f5ab2d9 100644 --- a/examples/dummyresource/resourcefactory.cpp +++ b/examples/dummyresource/resourcefactory.cpp | |||
@@ -126,13 +126,14 @@ class DummySynchronizer : public Sink::Synchronizer { | |||
126 | }); | 126 | }); |
127 | } | 127 | } |
128 | 128 | ||
129 | bool canReplay(const QByteArray &type, const QByteArray &key, const QByteArray &value) Q_DECL_OVERRIDE { return false; } | ||
130 | |||
129 | }; | 131 | }; |
130 | 132 | ||
131 | DummyResource::DummyResource(const Sink::ResourceContext &resourceContext, const QSharedPointer<Sink::Pipeline> &pipeline) | 133 | DummyResource::DummyResource(const Sink::ResourceContext &resourceContext, const QSharedPointer<Sink::Pipeline> &pipeline) |
132 | : Sink::GenericResource(resourceContext, pipeline) | 134 | : Sink::GenericResource(resourceContext, pipeline) |
133 | { | 135 | { |
134 | setupSynchronizer(QSharedPointer<DummySynchronizer>::create(resourceContext)); | 136 | setupSynchronizer(QSharedPointer<DummySynchronizer>::create(resourceContext)); |
135 | setupChangereplay(QSharedPointer<Sink::NullChangeReplay>::create(resourceContext)); | ||
136 | setupPreprocessors(ENTITY_TYPE_MAIL, | 137 | setupPreprocessors(ENTITY_TYPE_MAIL, |
137 | QVector<Sink::Preprocessor*>() << new MailPropertyExtractor); | 138 | QVector<Sink::Preprocessor*>() << new MailPropertyExtractor); |
138 | setupPreprocessors(ENTITY_TYPE_FOLDER, | 139 | setupPreprocessors(ENTITY_TYPE_FOLDER, |
diff --git a/examples/imapresource/imapresource.cpp b/examples/imapresource/imapresource.cpp index 2b9659e..8fc7702 100644 --- a/examples/imapresource/imapresource.cpp +++ b/examples/imapresource/imapresource.cpp | |||
@@ -28,7 +28,6 @@ | |||
28 | #include "definitions.h" | 28 | #include "definitions.h" |
29 | #include "inspection.h" | 29 | #include "inspection.h" |
30 | #include "synchronizer.h" | 30 | #include "synchronizer.h" |
31 | #include "sourcewriteback.h" | ||
32 | #include "remoteidmap.h" | 31 | #include "remoteidmap.h" |
33 | #include "query.h" | 32 | #include "query.h" |
34 | 33 | ||
@@ -393,22 +392,6 @@ public: | |||
393 | return KAsync::error<void>("Nothing to do"); | 392 | return KAsync::error<void>("Nothing to do"); |
394 | } | 393 | } |
395 | 394 | ||
396 | public: | ||
397 | QString mServer; | ||
398 | int mPort; | ||
399 | QString mUser; | ||
400 | QString mPassword; | ||
401 | QByteArray mResourceInstanceIdentifier; | ||
402 | }; | ||
403 | |||
404 | class ImapWriteback : public Sink::SourceWriteBack | ||
405 | { | ||
406 | public: | ||
407 | ImapWriteback(const ResourceContext &resourceContext) : Sink::SourceWriteBack(resourceContext) | ||
408 | { | ||
409 | |||
410 | } | ||
411 | |||
412 | KAsync::Job<QByteArray> replay(const ApplicationDomain::Mail &mail, Sink::Operation operation, const QByteArray &oldRemoteId, const QList<QByteArray> &changedProperties) Q_DECL_OVERRIDE | 395 | KAsync::Job<QByteArray> replay(const ApplicationDomain::Mail &mail, Sink::Operation operation, const QByteArray &oldRemoteId, const QList<QByteArray> &changedProperties) Q_DECL_OVERRIDE |
413 | { | 396 | { |
414 | auto imap = QSharedPointer<ImapServerProxy>::create(mServer, mPort); | 397 | auto imap = QSharedPointer<ImapServerProxy>::create(mServer, mPort); |
@@ -595,12 +578,6 @@ ImapResource::ImapResource(const ResourceContext &resourceContext, const QShared | |||
595 | synchronizer->mUser = mUser; | 578 | synchronizer->mUser = mUser; |
596 | synchronizer->mPassword = mPassword; | 579 | synchronizer->mPassword = mPassword; |
597 | setupSynchronizer(synchronizer); | 580 | setupSynchronizer(synchronizer); |
598 | auto changereplay = QSharedPointer<ImapWriteback>::create(resourceContext); | ||
599 | changereplay->mServer = mServer; | ||
600 | changereplay->mPort = mPort; | ||
601 | changereplay->mUser = mUser; | ||
602 | changereplay->mPassword = mPassword; | ||
603 | setupChangereplay(changereplay); | ||
604 | 581 | ||
605 | setupPreprocessors(ENTITY_TYPE_MAIL, QVector<Sink::Preprocessor*>() << new SpecialPurposeProcessor(resourceContext.resourceType, resourceContext.instanceId()) << new MimeMessageMover << new MailPropertyExtractor); | 582 | setupPreprocessors(ENTITY_TYPE_MAIL, QVector<Sink::Preprocessor*>() << new SpecialPurposeProcessor(resourceContext.resourceType, resourceContext.instanceId()) << new MimeMessageMover << new MailPropertyExtractor); |
606 | setupPreprocessors(ENTITY_TYPE_FOLDER, QVector<Sink::Preprocessor*>()); | 583 | setupPreprocessors(ENTITY_TYPE_FOLDER, QVector<Sink::Preprocessor*>()); |
diff --git a/examples/maildirresource/maildirresource.cpp b/examples/maildirresource/maildirresource.cpp index fc77315..fed18c4 100644 --- a/examples/maildirresource/maildirresource.cpp +++ b/examples/maildirresource/maildirresource.cpp | |||
@@ -28,7 +28,6 @@ | |||
28 | #include "libmaildir/maildir.h" | 28 | #include "libmaildir/maildir.h" |
29 | #include "inspection.h" | 29 | #include "inspection.h" |
30 | #include "synchronizer.h" | 30 | #include "synchronizer.h" |
31 | #include "sourcewriteback.h" | ||
32 | 31 | ||
33 | #include "facadefactory.h" | 32 | #include "facadefactory.h" |
34 | #include "adaptorfactoryregistry.h" | 33 | #include "adaptorfactoryregistry.h" |
@@ -383,18 +382,6 @@ public: | |||
383 | return job; | 382 | return job; |
384 | } | 383 | } |
385 | 384 | ||
386 | public: | ||
387 | QString mMaildirPath; | ||
388 | }; | ||
389 | |||
390 | class MaildirWriteback : public Sink::SourceWriteBack | ||
391 | { | ||
392 | public: | ||
393 | MaildirWriteback(const Sink::ResourceContext &resourceContext) : Sink::SourceWriteBack(resourceContext) | ||
394 | { | ||
395 | |||
396 | } | ||
397 | |||
398 | KAsync::Job<QByteArray> replay(const ApplicationDomain::Mail &mail, Sink::Operation operation, const QByteArray &oldRemoteId, const QList<QByteArray> &changedProperties) Q_DECL_OVERRIDE | 385 | KAsync::Job<QByteArray> replay(const ApplicationDomain::Mail &mail, Sink::Operation operation, const QByteArray &oldRemoteId, const QList<QByteArray> &changedProperties) Q_DECL_OVERRIDE |
399 | { | 386 | { |
400 | if (operation == Sink::Operation_Creation) { | 387 | if (operation == Sink::Operation_Creation) { |
@@ -453,9 +440,6 @@ MaildirResource::MaildirResource(const Sink::ResourceContext &resourceContext, c | |||
453 | auto synchronizer = QSharedPointer<MaildirSynchronizer>::create(resourceContext); | 440 | auto synchronizer = QSharedPointer<MaildirSynchronizer>::create(resourceContext); |
454 | synchronizer->mMaildirPath = mMaildirPath; | 441 | synchronizer->mMaildirPath = mMaildirPath; |
455 | setupSynchronizer(synchronizer); | 442 | setupSynchronizer(synchronizer); |
456 | auto changereplay = QSharedPointer<MaildirWriteback>::create(resourceContext); | ||
457 | changereplay->mMaildirPath = mMaildirPath; | ||
458 | setupChangereplay(changereplay); | ||
459 | 443 | ||
460 | setupPreprocessors(ENTITY_TYPE_MAIL, QVector<Sink::Preprocessor*>() << new SpecialPurposeProcessor(resourceContext.resourceType, resourceContext.instanceId()) << new MaildirMimeMessageMover(resourceContext.instanceId(), mMaildirPath) << new MaildirMailPropertyExtractor); | 444 | setupPreprocessors(ENTITY_TYPE_MAIL, QVector<Sink::Preprocessor*>() << new SpecialPurposeProcessor(resourceContext.resourceType, resourceContext.instanceId()) << new MaildirMimeMessageMover(resourceContext.instanceId(), mMaildirPath) << new MaildirMailPropertyExtractor); |
461 | setupPreprocessors(ENTITY_TYPE_FOLDER, QVector<Sink::Preprocessor*>() << new FolderPreprocessor(mMaildirPath)); | 445 | setupPreprocessors(ENTITY_TYPE_FOLDER, QVector<Sink::Preprocessor*>() << new FolderPreprocessor(mMaildirPath)); |
diff --git a/examples/mailtransportresource/mailtransportresource.cpp b/examples/mailtransportresource/mailtransportresource.cpp index 3fe551a..0688af6 100644 --- a/examples/mailtransportresource/mailtransportresource.cpp +++ b/examples/mailtransportresource/mailtransportresource.cpp | |||
@@ -23,7 +23,6 @@ | |||
23 | #include "resourceconfig.h" | 23 | #include "resourceconfig.h" |
24 | #include "definitions.h" | 24 | #include "definitions.h" |
25 | #include "domainadaptor.h" | 25 | #include "domainadaptor.h" |
26 | #include "sourcewriteback.h" | ||
27 | #include <QDir> | 26 | #include <QDir> |
28 | #include <QFileInfo> | 27 | #include <QFileInfo> |
29 | #include <QSettings> | 28 | #include <QSettings> |
@@ -45,30 +44,6 @@ SINK_DEBUG_AREA("mailtransportresource") | |||
45 | 44 | ||
46 | using namespace Sink; | 45 | using namespace Sink; |
47 | 46 | ||
48 | //TODO fold into synchronizer | ||
49 | class MailtransportWriteback : public Sink::SourceWriteBack | ||
50 | { | ||
51 | public: | ||
52 | MailtransportWriteback(const Sink::ResourceContext &resourceContext) : Sink::SourceWriteBack(resourceContext) | ||
53 | { | ||
54 | |||
55 | } | ||
56 | |||
57 | KAsync::Job<QByteArray> replay(const ApplicationDomain::Mail &mail, Sink::Operation operation, const QByteArray &oldRemoteId, const QList<QByteArray> &changedProperties) Q_DECL_OVERRIDE | ||
58 | { | ||
59 | if (operation == Sink::Operation_Creation) { | ||
60 | SinkTrace() << "Dispatching message."; | ||
61 | // return send(mail, mSettings); | ||
62 | } else if (operation == Sink::Operation_Removal) { | ||
63 | } else if (operation == Sink::Operation_Modification) { | ||
64 | } | ||
65 | return KAsync::null<QByteArray>(); | ||
66 | } | ||
67 | |||
68 | public: | ||
69 | MailtransportResource::Settings mSettings; | ||
70 | }; | ||
71 | |||
72 | class MailtransportSynchronizer : public Sink::Synchronizer { | 47 | class MailtransportSynchronizer : public Sink::Synchronizer { |
73 | public: | 48 | public: |
74 | MailtransportSynchronizer(const Sink::ResourceContext &resourceContext) | 49 | MailtransportSynchronizer(const Sink::ResourceContext &resourceContext) |
@@ -134,6 +109,17 @@ public: | |||
134 | }); | 109 | }); |
135 | } | 110 | } |
136 | 111 | ||
112 | KAsync::Job<QByteArray> replay(const ApplicationDomain::Mail &mail, Sink::Operation operation, const QByteArray &oldRemoteId, const QList<QByteArray> &changedProperties) Q_DECL_OVERRIDE | ||
113 | { | ||
114 | if (operation == Sink::Operation_Creation) { | ||
115 | SinkTrace() << "Dispatching message."; | ||
116 | // return send(mail, mSettings); | ||
117 | } else if (operation == Sink::Operation_Removal) { | ||
118 | } else if (operation == Sink::Operation_Modification) { | ||
119 | } | ||
120 | return KAsync::null<QByteArray>(); | ||
121 | } | ||
122 | |||
137 | public: | 123 | public: |
138 | QByteArray mResourceInstanceIdentifier; | 124 | QByteArray mResourceInstanceIdentifier; |
139 | MailtransportResource::Settings mSettings; | 125 | MailtransportResource::Settings mSettings; |
@@ -154,10 +140,6 @@ MailtransportResource::MailtransportResource(const Sink::ResourceContext &resour | |||
154 | synchronizer->mSettings = mSettings; | 140 | synchronizer->mSettings = mSettings; |
155 | setupSynchronizer(synchronizer); | 141 | setupSynchronizer(synchronizer); |
156 | 142 | ||
157 | auto changereplay = QSharedPointer<MailtransportWriteback>::create(resourceContext); | ||
158 | changereplay->mSettings = mSettings; | ||
159 | setupChangereplay(changereplay); | ||
160 | |||
161 | setupPreprocessors(ENTITY_TYPE_MAIL, QVector<Sink::Preprocessor*>() << new MimeMessageMover << new MailPropertyExtractor); | 143 | setupPreprocessors(ENTITY_TYPE_MAIL, QVector<Sink::Preprocessor*>() << new MimeMessageMover << new MailPropertyExtractor); |
162 | } | 144 | } |
163 | 145 | ||