1 files changed, 107 insertions, 62 deletions

diff --git a/common/pipeline.cpp b/common/pipeline.cpp
index 887b6b3..019784e 100644
--- a/common/pipeline.cpp
+++ b/common/pipeline.cpp
@@ -175,13 +175,14 @@ KAsync::Job<qint64> Pipeline::newEntity(void const *command, size_t size)
     auto o = Sink::ApplicationDomain::ApplicationDomainType{d->resourceContext.instanceId(), key, revision, memoryAdaptor};
     o.setChangedProperties(o.availableProperties().toSet());
 
-    auto preprocess = [&, this](ApplicationDomain::ApplicationDomainType &newEntity) {
-        foreach (const auto &processor, d->processors[bufferType]) {
-            processor->newEntity(newEntity);
-        }
-    };
+    auto newEntity = *ApplicationDomain::ApplicationDomainType::getInMemoryRepresentation<ApplicationDomain::ApplicationDomainType>(o, o.availableProperties());
+    newEntity.setChangedProperties(newEntity.availableProperties().toSet());
+
+    foreach (const auto &processor, d->processors[bufferType]) {
+        processor->newEntity(newEntity);
+    }
 
-    if (!d->entityStore.add(bufferType, o, replayToSource, preprocess)) {
+    if (!d->entityStore.add(bufferType, o, replayToSource)) {
         return KAsync::error<qint64>(0);
     }
 
@@ -195,6 +196,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,65 +254,71 @@ 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());
+    foreach (const auto &processor, d->processors[bufferType]) {
+        bool exitLoop = false;
+        const auto result = processor->processModification(Preprocessor::Modification, current, newEntity);
+        switch (result.action) {
+            case Preprocessor::MoveToResource:
+                isMove = true;
+                exitLoop = true;
+                break;
+            case Preprocessor::CopyToResource:
+                isMove = true;
+                exitLoop = true;
+                break;
+            case Preprocessor::DropModification:
+                SinkTraceCtx(d->logCtx) << "Dropping modification";
+                return KAsync::error<qint64>(0);
+            default:
+                break;
         }
-        newEntity.setResource(targetResource);
-        newEntity.setChangedProperties(newEntity.availableProperties().toSet());
+        if (exitLoop) {
+            break;
+        }
+    }
 
-        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) {
-                    startTransaction();
-                    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();
-                    commit();
+    //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();
-            }
-        });
-        job.exec();
-        return KAsync::value<qint64>(0);
+            })
+            .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);
     }
 
@@ -331,14 +343,14 @@ KAsync::Job<qint64> Pipeline::deletedEntity(void const *command, size_t size)
     const QByteArray key = QByteArray(reinterpret_cast<char const *>(deleteEntity->entityId()->Data()), deleteEntity->entityId()->size());
     SinkTraceCtx(d->logCtx) << "Deleted Entity. Type: " << bufferType << "uid: "<< key << " replayToSource: " << replayToSource;
 
-    auto preprocess = [&, this](const ApplicationDomain::ApplicationDomainType &oldEntity) {
-        foreach (const auto &processor, d->processors[bufferType]) {
-            processor->deletedEntity(oldEntity);
-        }
-    };
+    const auto current = d->entityStore.readLatest(bufferType, key);
+
+    foreach (const auto &processor, d->processors[bufferType]) {
+        processor->deletedEntity(current);
+    }
 
     d->revisionChanged = true;
-    if (!d->entityStore.remove(bufferType, key, replayToSource, preprocess)) {
+    if (!d->entityStore.remove(bufferType, current, replayToSource)) {
         return KAsync::error<qint64>(0);
     }
 
@@ -385,6 +397,39 @@ void Preprocessor::finalizeBatch()
 {
 }
 
+void Preprocessor::newEntity(ApplicationDomain::ApplicationDomainType &newEntity)
+{
+
+}
+
+void Preprocessor::modifiedEntity(const ApplicationDomain::ApplicationDomainType &oldEntity, ApplicationDomain::ApplicationDomainType &newEntity)
+{
+
+}
+
+void Preprocessor::deletedEntity(const ApplicationDomain::ApplicationDomainType &oldEntity)
+{
+
+}
+
+Preprocessor::Result Preprocessor::processModification(Type type, const ApplicationDomain::ApplicationDomainType &current, ApplicationDomain::ApplicationDomainType &diff)
+{
+    switch(type) {
+        case Creation:
+            newEntity(diff);
+            return {NoAction};
+        case Modification:
+            modifiedEntity(current, diff);
+            return {NoAction};
+        case Deletion:
+            deletedEntity(current);
+            return {NoAction};
+        default:
+            break;
+    }
+    return {NoAction};
+}
+
 QByteArray Preprocessor::resourceInstanceIdentifier() const
 {
     return d->resourceInstanceIdentifier;