From f6c3c144e60611d2da7ba7aa5b115affe92a57a4 Mon Sep 17 00:00:00 2001
From: Christian Mollekopf <chrigi_1@fastmail.fm>
Date: Tue, 11 Apr 2017 15:16:26 +0200
Subject: Move the preprocssing back out of entitystore into the pipeline.

This is where this really belongs, only the indexing is part of storage.
This is necessary so preprocessors can move entities as well.
---
 common/pipeline.cpp | 93 +++++++++++++++++++++++++----------------------------
 1 file changed, 43 insertions(+), 50 deletions(-)

(limited to 'common/pipeline.cpp')

diff --git a/common/pipeline.cpp b/common/pipeline.cpp
index 7f836c4..15ed5fc 100644
--- a/common/pipeline.cpp
+++ b/common/pipeline.cpp
@@ -195,6 +195,11 @@ struct CreateHelper {
     }
 };
 
+static KAsync::Job<void> create(const QByteArray &type, const ApplicationDomain::ApplicationDomainType &newEntity)
+{
+    return TypeHelper<CreateHelper>{type}.operator()<KAsync::Job<void>, const ApplicationDomain::ApplicationDomainType&>(newEntity);
+}
+
 KAsync::Job<qint64> Pipeline::modifiedEntity(void const *command, size_t size)
 {
     d->transactionItemCount++;
@@ -248,64 +253,52 @@ KAsync::Job<qint64> Pipeline::modifiedEntity(void const *command, size_t size)
         deletions = BufferUtils::fromVector(*modifyEntity->deletions());
     }
 
-    if (modifyEntity->targetResource()) {
-        auto isMove = modifyEntity->removeEntity();
-        auto targetResource = BufferUtils::extractBuffer(modifyEntity->targetResource());
-        auto changeset = diff.changedProperties();
-        const auto current = d->entityStore.readLatest(bufferType, diff.identifier());
-        if (current.identifier().isEmpty()) {
-            SinkWarningCtx(d->logCtx) << "Failed to read current version: " << diff.identifier();
-            return KAsync::error<qint64>(0);
-        }
+    const auto current = d->entityStore.readLatest(bufferType, diff.identifier());
+    if (current.identifier().isEmpty()) {
+        SinkWarningCtx(d->logCtx) << "Failed to read current version: " << diff.identifier();
+        return KAsync::error<qint64>(0);
+    }
 
-        auto newEntity = *ApplicationDomain::ApplicationDomainType::getInMemoryCopy<ApplicationDomain::ApplicationDomainType>(current, current.availableProperties());
+    auto newEntity = d->entityStore.applyDiff(bufferType, current, diff, deletions);
 
-        // Apply diff
-        for (const auto &property : changeset) {
-            const auto value = diff.getProperty(property);
-            if (value.isValid()) {
-                newEntity.setProperty(property, value);
-            }
-        }
+    bool isMove = false;
+    if (modifyEntity->targetResource()) {
+        isMove = modifyEntity->removeEntity();
+        newEntity.setResource(BufferUtils::extractBuffer(modifyEntity->targetResource()));
+    }
 
-        // Remove deletions
-        for (const auto &property : deletions) {
-            newEntity.setProperty(property, QVariant());
-        }
-        newEntity.setResource(targetResource);
-        newEntity.setChangedProperties(newEntity.availableProperties().toSet());
+    foreach (const auto &processor, d->processors[bufferType]) {
+        processor->modifiedEntity(current, newEntity);
+    }
 
-        SinkTraceCtx(d->logCtx) << "Moving entity to new resource " << newEntity.identifier() << newEntity.resourceInstanceIdentifier() << targetResource;
-        auto job = TypeHelper<CreateHelper>{bufferType}.operator()<KAsync::Job<void>, ApplicationDomain::ApplicationDomainType&>(newEntity);
-        job = job.then([this, current, isMove, targetResource, bufferType](const KAsync::Error &error) {
-            if (!error) {
-                SinkTraceCtx(d->logCtx) << "Move of " << current.identifier() << "was successfull";
-                if (isMove) {
-                    flatbuffers::FlatBufferBuilder fbb;
-                    auto entityId = fbb.CreateString(current.identifier());
-                    auto type = fbb.CreateString(bufferType);
-                    auto location = Sink::Commands::CreateDeleteEntity(fbb, current.revision(), entityId, type, true);
-                    Sink::Commands::FinishDeleteEntityBuffer(fbb, location);
-                    const auto data = BufferUtils::extractBuffer(fbb);
-                    deletedEntity(data, data.size()).exec();
+    //The entity is either being copied or moved
+    if (newEntity.resourceInstanceIdentifier() != d->resourceContext.resourceInstanceIdentifier) {
+        SinkTraceCtx(d->logCtx) << "Moving entity to new resource " << newEntity.identifier() << newEntity.resourceInstanceIdentifier();
+        newEntity.setChangedProperties(newEntity.availableProperties().toSet());
+        return create(bufferType, newEntity)
+            .then([=](const KAsync::Error &error) {
+                if (!error) {
+                    SinkTraceCtx(d->logCtx) << "Move of " << current.identifier() << "was successfull";
+                    if (isMove) {
+                        flatbuffers::FlatBufferBuilder fbb;
+                        auto entityId = fbb.CreateString(current.identifier());
+                        auto type = fbb.CreateString(bufferType);
+                        auto location = Sink::Commands::CreateDeleteEntity(fbb, current.revision(), entityId, type, true);
+                        Sink::Commands::FinishDeleteEntityBuffer(fbb, location);
+                        const auto data = BufferUtils::extractBuffer(fbb);
+                        deletedEntity(data, data.size()).exec();
+                    }
+                } else {
+                    SinkErrorCtx(d->logCtx) << "Failed to move entity " << newEntity.identifier() << " to resource " << newEntity.resourceInstanceIdentifier();
                 }
-            } else {
-                SinkErrorCtx(d->logCtx) << "Failed to move entity " << targetResource << " to resource " << current.identifier();
-            }
-        });
-        return job.then([this] {
-            return d->entityStore.maxRevision();
-        });
+            })
+            .then([this] {
+                return d->entityStore.maxRevision();
+            });
     }
 
-    auto preprocess = [&, this](const ApplicationDomain::ApplicationDomainType &oldEntity, ApplicationDomain::ApplicationDomainType &newEntity) {
-        foreach (const auto &processor, d->processors[bufferType]) {
-            processor->modifiedEntity(oldEntity, newEntity);
-        }
-    };
-
     d->revisionChanged = true;
-    if (!d->entityStore.modify(bufferType, diff, deletions, replayToSource, preprocess)) {
+    if (!d->entityStore.modify(bufferType, current, newEntity, replayToSource)) {
         return KAsync::error<qint64>(0);
     }
 
-- 
cgit v1.2.3