1 files changed, 119 insertions, 128 deletions

diff --git a/common/pipeline.cpp b/common/pipeline.cpp
index 35e582b..65a2f5b 100644
--- a/common/pipeline.cpp
+++ b/common/pipeline.cpp
@@ -40,21 +40,18 @@
 #undef DEBUG_AREA
 #define DEBUG_AREA "resource.pipeline"
 
-namespace Sink
-{
+namespace Sink {
 
 class Pipeline::Private
 {
 public:
-    Private(const QString &resourceName)
-        : storage(Sink::storageLocation(), resourceName, Storage::ReadWrite),
-        revisionChanged(false)
+    Private(const QString &resourceName) : storage(Sink::storageLocation(), resourceName, Storage::ReadWrite), revisionChanged(false)
     {
     }
 
     Storage storage;
     Storage::Transaction transaction;
-    QHash<QString, QVector<Preprocessor *> > processors;
+    QHash<QString, QVector<Preprocessor *>> processors;
     QHash<QString, DomainTypeAdaptorFactoryInterface::Ptr> adaptorFactory;
     bool revisionChanged;
     void storeNewRevision(qint64 newRevision, const flatbuffers::FlatBufferBuilder &fbb, const QByteArray &bufferType, const QByteArray &uid);
@@ -64,20 +61,16 @@ public:
 
 void Pipeline::Private::storeNewRevision(qint64 newRevision, const flatbuffers::FlatBufferBuilder &fbb, const QByteArray &bufferType, const QByteArray &uid)
 {
-    Storage::mainDatabase(transaction, bufferType).write(Storage::assembleKey(uid, newRevision), BufferUtils::extractBuffer(fbb),
-        [uid, newRevision](const Storage::Error &error) {
-            Warning() << "Failed to write entity" << uid << newRevision;
-        }
-    );
+    Storage::mainDatabase(transaction, bufferType)
+        .write(Storage::assembleKey(uid, newRevision), BufferUtils::extractBuffer(fbb),
+            [uid, newRevision](const Storage::Error &error) { Warning() << "Failed to write entity" << uid << newRevision; });
     revisionChanged = true;
     Storage::setMaxRevision(transaction, newRevision);
     Storage::recordRevision(transaction, newRevision, uid, bufferType);
 }
 
 
-Pipeline::Pipeline(const QString &resourceName, QObject *parent)
-    : QObject(parent),
-      d(new Private(resourceName))
+Pipeline::Pipeline(const QString &resourceName, QObject *parent) : QObject(parent), d(new Private(resourceName))
 {
 }
 
@@ -98,8 +91,8 @@ void Pipeline::setAdaptorFactory(const QString &entityType, DomainTypeAdaptorFac
 
 void Pipeline::startTransaction()
 {
-    //TODO call for all types
-    //But avoid doing it during cleanup
+    // TODO call for all types
+    // But avoid doing it during cleanup
     // for (auto processor : d->processors[bufferType]) {
     //     processor->startBatch();
     // }
@@ -114,14 +107,15 @@ void Pipeline::startTransaction()
 
 void Pipeline::commit()
 {
-    //TODO call for all types
-    //But avoid doing it during cleanup
+    // TODO call for all types
+    // But avoid doing it during cleanup
     // for (auto processor : d->processors[bufferType]) {
     //     processor->finalize();
     // }
     const auto revision = Storage::maxRevision(d->transaction);
     const auto elapsed = d->transactionTime.elapsed();
-    Trace() << "Committing revision: " << revision << ":" << d->transactionItemCount << " items in: " << Log::TraceTime(elapsed) << " " << (double)elapsed/(double)qMax(d->transactionItemCount, 1) << "[ms/item]";
+    Trace() << "Committing revision: " << revision << ":" << d->transactionItemCount << " items in: " << Log::TraceTime(elapsed) << " "
+            << (double)elapsed / (double)qMax(d->transactionItemCount, 1) << "[ms/item]";
     if (d->transaction) {
         d->transaction.commit();
     }
@@ -157,7 +151,7 @@ KAsync::Job<qint64> Pipeline::newEntity(void const *command, size_t size)
     auto createEntity = Commands::GetCreateEntity(command);
 
     const bool replayToSource = createEntity->replayToSource();
-    const QByteArray bufferType = QByteArray(reinterpret_cast<char const*>(createEntity->domainType()->Data()), createEntity->domainType()->size());
+    const QByteArray bufferType = QByteArray(reinterpret_cast<char const *>(createEntity->domainType()->Data()), createEntity->domainType()->size());
     {
         flatbuffers::Verifier verifyer(reinterpret_cast<const uint8_t *>(createEntity->delta()->Data()), createEntity->delta()->size());
         if (!VerifyEntityBuffer(verifyer)) {
@@ -173,7 +167,7 @@ KAsync::Job<qint64> Pipeline::newEntity(void const *command, size_t size)
 
     QByteArray key;
     if (createEntity->entityId()) {
-        key = QByteArray(reinterpret_cast<char const*>(createEntity->entityId()->Data()), createEntity->entityId()->size());
+        key = QByteArray(reinterpret_cast<char const *>(createEntity->entityId()->Data()), createEntity->entityId()->size());
         if (Storage::mainDatabase(d->transaction, bufferType).contains(key)) {
             ErrorMsg() << "An entity with this id already exists: " << key;
             return KAsync::error<qint64>(0);
@@ -186,7 +180,7 @@ KAsync::Job<qint64> Pipeline::newEntity(void const *command, size_t size)
     Q_ASSERT(!key.isEmpty());
     const qint64 newRevision = Storage::maxRevision(d->transaction) + 1;
 
-    //Add metadata buffer
+    // Add metadata buffer
     flatbuffers::FlatBufferBuilder metadataFbb;
     auto metadataBuilder = MetadataBuilder(metadataFbb);
     metadataBuilder.add_revision(newRevision);
@@ -196,7 +190,8 @@ KAsync::Job<qint64> Pipeline::newEntity(void const *command, size_t size)
     FinishMetadataBuffer(metadataFbb, metadataBuffer);
 
     flatbuffers::FlatBufferBuilder fbb;
-    EntityBuffer::assembleEntityBuffer(fbb, metadataFbb.GetBufferPointer(), metadataFbb.GetSize(), entity->resource()->Data(), entity->resource()->size(), entity->local()->Data(), entity->local()->size());
+    EntityBuffer::assembleEntityBuffer(
+        fbb, metadataFbb.GetBufferPointer(), metadataFbb.GetSize(), entity->resource()->Data(), entity->resource()->size(), entity->local()->Data(), entity->local()->size());
 
     d->storeNewRevision(newRevision, fbb, bufferType, key);
 
@@ -207,20 +202,19 @@ KAsync::Job<qint64> Pipeline::newEntity(void const *command, size_t size)
     }
 
     Log() << "Pipeline: wrote entity: " << key << newRevision << bufferType;
-    Storage::mainDatabase(d->transaction, bufferType).scan(Storage::assembleKey(key, newRevision), [this, bufferType, newRevision, adaptorFactory, key](const QByteArray &, const QByteArray &value) -> bool {
-        auto entity = GetEntity(value);
-        Q_ASSERT(entity->resource() || entity->local());
-        auto adaptor = adaptorFactory->createAdaptor(*entity);
-        for (auto processor : d->processors[bufferType]) {
-            processor->newEntity(key, newRevision, *adaptor, d->transaction);
-        }
-        return false;
-    }, [this](const Storage::Error &error) {
-        ErrorMsg() << "Failed to find value in pipeline: " << error.message;
-    });
-    return KAsync::start<qint64>([newRevision](){
-        return newRevision;
-    });
+    Storage::mainDatabase(d->transaction, bufferType)
+        .scan(Storage::assembleKey(key, newRevision),
+            [this, bufferType, newRevision, adaptorFactory, key](const QByteArray &, const QByteArray &value) -> bool {
+                auto entity = GetEntity(value);
+                Q_ASSERT(entity->resource() || entity->local());
+                auto adaptor = adaptorFactory->createAdaptor(*entity);
+                for (auto processor : d->processors[bufferType]) {
+                    processor->newEntity(key, newRevision, *adaptor, d->transaction);
+                }
+                return false;
+            },
+            [this](const Storage::Error &error) { ErrorMsg() << "Failed to find value in pipeline: " << error.message; });
+    return KAsync::start<qint64>([newRevision]() { return newRevision; });
 }
 
 KAsync::Job<qint64> Pipeline::modifiedEntity(void const *command, size_t size)
@@ -242,9 +236,9 @@ KAsync::Job<qint64> Pipeline::modifiedEntity(void const *command, size_t size)
 
     const qint64 baseRevision = modifyEntity->revision();
     const bool replayToSource = modifyEntity->replayToSource();
-    //TODO rename modifyEntity->domainType to bufferType
-    const QByteArray bufferType = QByteArray(reinterpret_cast<char const*>(modifyEntity->domainType()->Data()), modifyEntity->domainType()->size());
-    const QByteArray key = QByteArray(reinterpret_cast<char const*>(modifyEntity->entityId()->Data()), modifyEntity->entityId()->size());
+    // TODO rename modifyEntity->domainType to bufferType
+    const QByteArray bufferType = QByteArray(reinterpret_cast<char const *>(modifyEntity->domainType()->Data()), modifyEntity->domainType()->size());
+    const QByteArray key = QByteArray(reinterpret_cast<char const *>(modifyEntity->entityId()->Data()), modifyEntity->entityId()->size());
     if (bufferType.isEmpty() || key.isEmpty()) {
         Warning() << "entity type or key " << bufferType << key;
         return KAsync::error<qint64>(0);
@@ -257,7 +251,7 @@ KAsync::Job<qint64> Pipeline::modifiedEntity(void const *command, size_t size)
         }
     }
 
-    //TODO use only readPropertyMapper and writePropertyMapper
+    // TODO use only readPropertyMapper and writePropertyMapper
     auto adaptorFactory = d->adaptorFactory.value(bufferType);
     if (!adaptorFactory) {
         Warning() << "no adaptor factory for type " << bufferType;
@@ -269,30 +263,30 @@ KAsync::Job<qint64> Pipeline::modifiedEntity(void const *command, size_t size)
     auto diff = adaptorFactory->createAdaptor(*diffEntity);
 
     QSharedPointer<ApplicationDomain::BufferAdaptor> current;
-    Storage::mainDatabase(d->transaction, bufferType).findLatest(key, [&current, adaptorFactory](const QByteArray &key, const QByteArray &data) -> bool {
-        EntityBuffer buffer(const_cast<const char *>(data.data()), data.size());
-        if (!buffer.isValid()) {
-            Warning() << "Read invalid buffer from disk";
-        } else {
-            current = adaptorFactory->createAdaptor(buffer.entity());
-        }
-        return false;
-    },
-    [baseRevision](const Storage::Error &error) {
-        Warning() << "Failed to read old revision from storage: " << error.message << "Revision: " << baseRevision;
-    });
+    Storage::mainDatabase(d->transaction, bufferType)
+        .findLatest(key,
+            [&current, adaptorFactory](const QByteArray &key, const QByteArray &data) -> bool {
+                EntityBuffer buffer(const_cast<const char *>(data.data()), data.size());
+                if (!buffer.isValid()) {
+                    Warning() << "Read invalid buffer from disk";
+                } else {
+                    current = adaptorFactory->createAdaptor(buffer.entity());
+                }
+                return false;
+            },
+            [baseRevision](const Storage::Error &error) { Warning() << "Failed to read old revision from storage: " << error.message << "Revision: " << baseRevision; });
 
     if (!current) {
         Warning() << "Failed to read local value " << key;
         return KAsync::error<qint64>(0);
     }
 
-    //resource and uid don't matter at this point
+    // resource and uid don't matter at this point
     const ApplicationDomain::ApplicationDomainType existingObject("", "", newRevision, current);
     auto newObject = ApplicationDomain::ApplicationDomainType::getInMemoryRepresentation<ApplicationDomain::ApplicationDomainType>(existingObject);
 
-    //Apply diff
-    //FIXME only apply the properties that are available in the buffer
+    // Apply diff
+    // FIXME only apply the properties that are available in the buffer
     Trace() << "Applying changed properties: " << diff->availableProperties();
     QSet<QByteArray> changeset;
     for (const auto &property : diff->availableProperties()) {
@@ -302,17 +296,17 @@ KAsync::Job<qint64> Pipeline::modifiedEntity(void const *command, size_t size)
             newObject->setProperty(property, value);
         }
     }
-    //Altough we only set some properties, we want all to be serialized
+    // Altough we only set some properties, we want all to be serialized
     newObject->setChangedProperties(changeset);
 
-    //Remove deletions
+    // Remove deletions
     if (modifyEntity->deletions()) {
         for (const auto &property : *modifyEntity->deletions()) {
             newObject->setProperty(BufferUtils::extractBuffer(property), QVariant());
         }
     }
 
-    //Add metadata buffer
+    // Add metadata buffer
     flatbuffers::FlatBufferBuilder metadataFbb;
     auto metadataBuilder = MetadataBuilder(metadataFbb);
     metadataBuilder.add_revision(newRevision);
@@ -326,22 +320,21 @@ KAsync::Job<qint64> Pipeline::modifiedEntity(void const *command, size_t size)
 
     d->storeNewRevision(newRevision, fbb, bufferType, key);
     Log() << "Pipeline: modified entity: " << key << newRevision << bufferType;
-    Storage::mainDatabase(d->transaction, bufferType).scan(Storage::assembleKey(key, newRevision), [this, bufferType, newRevision, adaptorFactory, current, key](const QByteArray &k, const QByteArray &value) -> bool {
-        if (value.isEmpty()) {
-            ErrorMsg() << "Read buffer is empty.";
-        }
-        auto entity = GetEntity(value.data());
-        auto newEntity = adaptorFactory->createAdaptor(*entity);
-        for (auto processor : d->processors[bufferType]) {
-            processor->modifiedEntity(key, newRevision, *current, *newEntity, d->transaction);
-        }
-        return false;
-    }, [this](const Storage::Error &error) {
-        ErrorMsg() << "Failed to find value in pipeline: " << error.message;
-    });
-    return KAsync::start<qint64>([newRevision](){
-        return newRevision;
-    });
+    Storage::mainDatabase(d->transaction, bufferType)
+        .scan(Storage::assembleKey(key, newRevision),
+            [this, bufferType, newRevision, adaptorFactory, current, key](const QByteArray &k, const QByteArray &value) -> bool {
+                if (value.isEmpty()) {
+                    ErrorMsg() << "Read buffer is empty.";
+                }
+                auto entity = GetEntity(value.data());
+                auto newEntity = adaptorFactory->createAdaptor(*entity);
+                for (auto processor : d->processors[bufferType]) {
+                    processor->modifiedEntity(key, newRevision, *current, *newEntity, d->transaction);
+                }
+                return false;
+            },
+            [this](const Storage::Error &error) { ErrorMsg() << "Failed to find value in pipeline: " << error.message; });
+    return KAsync::start<qint64>([newRevision]() { return newRevision; });
 }
 
 KAsync::Job<qint64> Pipeline::deletedEntity(void const *command, size_t size)
@@ -359,26 +352,25 @@ KAsync::Job<qint64> Pipeline::deletedEntity(void const *command, size_t size)
     auto deleteEntity = Commands::GetDeleteEntity(command);
 
     const bool replayToSource = deleteEntity->replayToSource();
-    const QByteArray bufferType = QByteArray(reinterpret_cast<char const*>(deleteEntity->domainType()->Data()), deleteEntity->domainType()->size());
-    const QByteArray key = QByteArray(reinterpret_cast<char const*>(deleteEntity->entityId()->Data()), deleteEntity->entityId()->size());
+    const QByteArray bufferType = QByteArray(reinterpret_cast<char const *>(deleteEntity->domainType()->Data()), deleteEntity->domainType()->size());
+    const QByteArray key = QByteArray(reinterpret_cast<char const *>(deleteEntity->entityId()->Data()), deleteEntity->entityId()->size());
 
     bool found = false;
     bool alreadyRemoved = false;
-    Storage::mainDatabase(d->transaction, bufferType).findLatest(key, [&found, &alreadyRemoved](const QByteArray &key, const QByteArray &data) -> bool {
-        auto entity = GetEntity(data.data());
-        if (entity && entity->metadata()) {
-            auto metadata = GetMetadata(entity->metadata()->Data());
-            found = true;
-            if (metadata->operation() == Operation_Removal) {
-                alreadyRemoved = true;
-            }
-
-        }
-        return false;
-    },
-    [](const Storage::Error &error) {
-        Warning() << "Failed to read old revision from storage: " << error.message;
-    });
+    Storage::mainDatabase(d->transaction, bufferType)
+        .findLatest(key,
+            [&found, &alreadyRemoved](const QByteArray &key, const QByteArray &data) -> bool {
+                auto entity = GetEntity(data.data());
+                if (entity && entity->metadata()) {
+                    auto metadata = GetMetadata(entity->metadata()->Data());
+                    found = true;
+                    if (metadata->operation() == Operation_Removal) {
+                        alreadyRemoved = true;
+                    }
+                }
+                return false;
+            },
+            [](const Storage::Error &error) { Warning() << "Failed to read old revision from storage: " << error.message; });
 
     if (!found) {
         Warning() << "Failed to find entity " << key;
@@ -391,7 +383,7 @@ KAsync::Job<qint64> Pipeline::deletedEntity(void const *command, size_t size)
 
     const qint64 newRevision = Storage::maxRevision(d->transaction) + 1;
 
-    //Add metadata buffer
+    // Add metadata buffer
     flatbuffers::FlatBufferBuilder metadataFbb;
     auto metadataBuilder = MetadataBuilder(metadataFbb);
     metadataBuilder.add_revision(newRevision);
@@ -410,28 +402,27 @@ KAsync::Job<qint64> Pipeline::deletedEntity(void const *command, size_t size)
     }
 
     QSharedPointer<ApplicationDomain::BufferAdaptor> current;
-    Storage::mainDatabase(d->transaction, bufferType).findLatest(key, [this, bufferType, newRevision, adaptorFactory, key, &current](const QByteArray &, const QByteArray &data) -> bool {
-        EntityBuffer buffer(const_cast<const char *>(data.data()), data.size());
-        if (!buffer.isValid()) {
-            Warning() << "Read invalid buffer from disk";
-        } else {
-            current = adaptorFactory->createAdaptor(buffer.entity());
-        }
-        return false;
-    }, [this](const Storage::Error &error) {
-        ErrorMsg() << "Failed to find value in pipeline: " << error.message;
-    });
+    Storage::mainDatabase(d->transaction, bufferType)
+        .findLatest(key,
+            [this, bufferType, newRevision, adaptorFactory, key, &current](const QByteArray &, const QByteArray &data) -> bool {
+                EntityBuffer buffer(const_cast<const char *>(data.data()), data.size());
+                if (!buffer.isValid()) {
+                    Warning() << "Read invalid buffer from disk";
+                } else {
+                    current = adaptorFactory->createAdaptor(buffer.entity());
+                }
+                return false;
+            },
+            [this](const Storage::Error &error) { ErrorMsg() << "Failed to find value in pipeline: " << error.message; });
 
     d->storeNewRevision(newRevision, fbb, bufferType, key);
-    Log() << "Pipeline: deleted entity: "<< newRevision;
+    Log() << "Pipeline: deleted entity: " << newRevision;
 
     for (auto processor : d->processors[bufferType]) {
         processor->deletedEntity(key, newRevision, *current, d->transaction);
     }
 
-    return KAsync::start<qint64>([newRevision](){
-        return newRevision;
-    });
+    return KAsync::start<qint64>([newRevision]() { return newRevision; });
 }
 
 void Pipeline::cleanupRevision(qint64 revision)
@@ -439,24 +430,25 @@ void Pipeline::cleanupRevision(qint64 revision)
     const auto uid = Storage::getUidFromRevision(d->transaction, revision);
     const auto bufferType = Storage::getTypeFromRevision(d->transaction, revision);
     Trace() << "Cleaning up revision " << revision << uid << bufferType;
-    Storage::mainDatabase(d->transaction, bufferType).scan(uid, [&](const QByteArray &key, const QByteArray &data) -> bool {
-        EntityBuffer buffer(const_cast<const char *>(data.data()), data.size());
-        if (!buffer.isValid()) {
-            Warning() << "Read invalid buffer from disk";
-        } else {
-            const auto metadata =  flatbuffers::GetRoot<Metadata>(buffer.metadataBuffer());
-            const qint64 rev = metadata->revision();
-            //Remove old revisions, and the current if the entity has already been removed
-            if (rev < revision || metadata->operation() == Operation_Removal) {
-                Storage::removeRevision(d->transaction, rev);
-                Storage::mainDatabase(d->transaction, bufferType).remove(key);
-            }
-        }
-
-        return true;
-    }, [](const Storage::Error &error) {
-        Warning() << "Error while reading: " << error.message;
-    }, true);
+    Storage::mainDatabase(d->transaction, bufferType)
+        .scan(uid,
+            [&](const QByteArray &key, const QByteArray &data) -> bool {
+                EntityBuffer buffer(const_cast<const char *>(data.data()), data.size());
+                if (!buffer.isValid()) {
+                    Warning() << "Read invalid buffer from disk";
+                } else {
+                    const auto metadata = flatbuffers::GetRoot<Metadata>(buffer.metadataBuffer());
+                    const qint64 rev = metadata->revision();
+                    // Remove old revisions, and the current if the entity has already been removed
+                    if (rev < revision || metadata->operation() == Operation_Removal) {
+                        Storage::removeRevision(d->transaction, rev);
+                        Storage::mainDatabase(d->transaction, bufferType).remove(key);
+                    }
+                }
+
+                return true;
+            },
+            [](const Storage::Error &error) { Warning() << "Error while reading: " << error.message; }, true);
     Storage::setCleanedUpRevision(d->transaction, revision);
 }
 
@@ -465,8 +457,7 @@ qint64 Pipeline::cleanedUpRevision()
     return Storage::cleanedUpRevision(d->transaction);
 }
 
-Preprocessor::Preprocessor()
-    : d(0)
+Preprocessor::Preprocessor() : d(0)
 {
 }