1 files changed, 45 insertions, 45 deletions

diff --git a/common/pipeline.cpp b/common/pipeline.cpp
index c6d5297..f1a4a32 100644
--- a/common/pipeline.cpp
+++ b/common/pipeline.cpp
@@ -38,8 +38,7 @@
 #include "definitions.h"
 #include "bufferutils.h"
 
-#undef DEBUG_AREA
-#define DEBUG_AREA "resource.pipeline"
+SINK_DEBUG_AREA("pipeline")
 
 namespace Sink {
 
@@ -52,7 +51,7 @@ public:
 
     Storage storage;
     Storage::Transaction transaction;
-    QHash<QString, QVector<Preprocessor *>> processors;
+    QHash<QString, QVector<QSharedPointer<Preprocessor>>> processors;
     bool revisionChanged;
     void storeNewRevision(qint64 newRevision, const flatbuffers::FlatBufferBuilder &fbb, const QByteArray &bufferType, const QByteArray &uid);
     QTime transactionTime;
@@ -63,10 +62,10 @@ public:
 
 void Pipeline::Private::storeNewRevision(qint64 newRevision, const flatbuffers::FlatBufferBuilder &fbb, const QByteArray &bufferType, const QByteArray &uid)
 {
-    Trace() << "Committing new revision: " << uid << newRevision;
+    SinkTrace() << "Committing new revision: " << 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; });
+            [uid, newRevision](const Storage::Error &error) { SinkWarning() << "Failed to write entity" << uid << newRevision; });
     revisionChanged = true;
     Storage::setMaxRevision(transaction, newRevision);
     Storage::recordRevision(transaction, newRevision, uid, bufferType);
@@ -79,15 +78,17 @@ Pipeline::Pipeline(const QString &resourceName, QObject *parent) : QObject(paren
 
 Pipeline::~Pipeline()
 {
-    delete d;
+    d->transaction = Storage::Transaction();
 }
 
 void Pipeline::setPreprocessors(const QString &entityType, const QVector<Preprocessor *> &processors)
 {
+    auto &list = d->processors[entityType];
+    list.clear();
     for (auto p : processors) {
         p->setup(d->resourceType, d->resourceInstanceIdentifier, this);
+        list.append(QSharedPointer<Preprocessor>(p));
     }
-    d->processors[entityType] = processors;
 }
 
 void Pipeline::setResourceType(const QByteArray &resourceType)
@@ -105,21 +106,21 @@ void Pipeline::startTransaction()
     if (d->transaction) {
         return;
     }
-    Trace() << "Starting transaction.";
+    SinkTrace() << "Starting transaction.";
     d->transactionTime.start();
     d->transactionItemCount = 0;
-    d->transaction = std::move(storage().createTransaction(Storage::ReadWrite, [](const Sink::Storage::Error &error) {
-        Warning() << error.message;
-    }));
+    d->transaction = storage().createTransaction(Storage::ReadWrite, [](const Sink::Storage::Error &error) {
+        SinkWarning() << error.message;
+    });
 
     //FIXME this is a temporary measure to recover from a failure to open the named databases correctly.
     //Once the actual problem is fixed it will be enough to simply crash if we open the wrong database (which we check in openDatabase already).
     //It seems like the validateNamedDatabase calls actually stops the mdb_put failures during sync...
     if (d->storage.exists()) {
         while (!d->transaction.validateNamedDatabases()) {
-            Warning() << "Opened an invalid transaction!!!!!!";
+            SinkWarning() << "Opened an invalid transaction!!!!!!";
             d->transaction = std::move(storage().createTransaction(Storage::ReadWrite, [](const Sink::Storage::Error &error) {
-                Warning() << error.message;
+                SinkWarning() << error.message;
             }));
         }
     }
@@ -139,7 +140,7 @@ void Pipeline::commit()
     }
     const auto revision = Storage::maxRevision(d->transaction);
     const auto elapsed = d->transactionTime.elapsed();
-    Log() << "Committing revision: " << revision << ":" << d->transactionItemCount << " items in: " << Log::TraceTime(elapsed) << " "
+    SinkLog() << "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();
@@ -168,7 +169,7 @@ KAsync::Job<qint64> Pipeline::newEntity(void const *command, size_t size)
     {
         flatbuffers::Verifier verifyer(reinterpret_cast<const uint8_t *>(command), size);
         if (!Commands::VerifyCreateEntityBuffer(verifyer)) {
-            Warning() << "invalid buffer, not a create entity buffer";
+            SinkWarning() << "invalid buffer, not a create entity buffer";
             return KAsync::error<qint64>(0);
         }
     }
@@ -180,7 +181,7 @@ KAsync::Job<qint64> Pipeline::newEntity(void const *command, size_t size)
     if (createEntity->entityId()) {
         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;
+            SinkError() << "An entity with this id already exists: " << key;
             return KAsync::error<qint64>(0);
         }
     }
@@ -188,31 +189,31 @@ KAsync::Job<qint64> Pipeline::newEntity(void const *command, size_t size)
     if (key.isEmpty()) {
         key = Sink::Storage::generateUid();
     }
-    Log() << "New Entity. Type: " << bufferType << "uid: "<< key << " replayToSource: " << replayToSource;
+    SinkLog() << "New Entity. Type: " << bufferType << "uid: "<< key << " replayToSource: " << replayToSource;
     Q_ASSERT(!key.isEmpty());
 
     {
         flatbuffers::Verifier verifyer(reinterpret_cast<const uint8_t *>(createEntity->delta()->Data()), createEntity->delta()->size());
         if (!VerifyEntityBuffer(verifyer)) {
-            Warning() << "invalid buffer, not an entity buffer";
+            SinkWarning() << "invalid buffer, not an entity buffer";
             return KAsync::error<qint64>(0);
         }
     }
     auto entity = GetEntity(createEntity->delta()->Data());
     if (!entity->resource()->size() && !entity->local()->size()) {
-        Warning() << "No local and no resource buffer while trying to create entity.";
+        SinkWarning() << "No local and no resource buffer while trying to create entity.";
         return KAsync::error<qint64>(0);
     }
 
     auto adaptorFactory = Sink::AdaptorFactoryRegistry::instance().getFactory(d->resourceType, bufferType);
     if (!adaptorFactory) {
-        Warning() << "no adaptor factory for type " << bufferType;
+        SinkWarning() << "no adaptor factory for type " << bufferType;
         return KAsync::error<qint64>(0);
     }
 
     auto adaptor = adaptorFactory->createAdaptor(*entity);
     auto memoryAdaptor = QSharedPointer<Sink::ApplicationDomain::MemoryBufferAdaptor>::create(*(adaptor), adaptor->availableProperties());
-    for (auto processor : d->processors[bufferType]) {
+    foreach (const auto &processor, d->processors[bufferType]) {
         processor->newEntity(key, Storage::maxRevision(d->transaction) + 1, *memoryAdaptor, d->transaction);
     }
     //The maxRevision may have changed meanwhile if the entity created sub-entities
@@ -242,7 +243,7 @@ KAsync::Job<qint64> Pipeline::modifiedEntity(void const *command, size_t size)
     {
         flatbuffers::Verifier verifyer(reinterpret_cast<const uint8_t *>(command), size);
         if (!Commands::VerifyModifyEntityBuffer(verifyer)) {
-            Warning() << "invalid buffer, not a modify entity buffer";
+            SinkWarning() << "invalid buffer, not a modify entity buffer";
             return KAsync::error<qint64>(0);
         }
     }
@@ -252,21 +253,21 @@ KAsync::Job<qint64> Pipeline::modifiedEntity(void const *command, size_t size)
     if (modifyEntity->modifiedProperties()) {
         changeset = BufferUtils::fromVector(*modifyEntity->modifiedProperties());
     } else {
-        Warning() << "No changeset available";
+        SinkWarning() << "No changeset available";
     }
     const qint64 baseRevision = modifyEntity->revision();
     const bool replayToSource = modifyEntity->replayToSource();
     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());
-    Log() << "Modified Entity. Type: " << bufferType << "uid: "<< key << " replayToSource: " << replayToSource;
+    SinkLog() << "Modified Entity. Type: " << bufferType << "uid: "<< key << " replayToSource: " << replayToSource;
     if (bufferType.isEmpty() || key.isEmpty()) {
-        Warning() << "entity type or key " << bufferType << key;
+        SinkWarning() << "entity type or key " << bufferType << key;
         return KAsync::error<qint64>(0);
     }
     {
         flatbuffers::Verifier verifyer(reinterpret_cast<const uint8_t *>(modifyEntity->delta()->Data()), modifyEntity->delta()->size());
         if (!VerifyEntityBuffer(verifyer)) {
-            Warning() << "invalid buffer, not an entity buffer";
+            SinkWarning() << "invalid buffer, not an entity buffer";
             return KAsync::error<qint64>(0);
         }
     }
@@ -274,7 +275,7 @@ KAsync::Job<qint64> Pipeline::modifiedEntity(void const *command, size_t size)
     // TODO use only readPropertyMapper and writePropertyMapper
     auto adaptorFactory = Sink::AdaptorFactoryRegistry::instance().getFactory(d->resourceType, bufferType);
     if (!adaptorFactory) {
-        Warning() << "no adaptor factory for type " << bufferType;
+        SinkWarning() << "no adaptor factory for type " << bufferType;
         return KAsync::error<qint64>(0);
     }
 
@@ -288,16 +289,16 @@ KAsync::Job<qint64> Pipeline::modifiedEntity(void const *command, size_t size)
             [&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";
+                    SinkWarning() << "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; });
+            [baseRevision](const Storage::Error &error) { SinkWarning() << "Failed to read old revision from storage: " << error.message << "Revision: " << baseRevision; });
 
     if (!current) {
-        Warning() << "Failed to read local value " << key;
+        SinkWarning() << "Failed to read local value " << key;
         return KAsync::error<qint64>(0);
     }
 
@@ -305,7 +306,7 @@ KAsync::Job<qint64> Pipeline::modifiedEntity(void const *command, size_t size)
 
     // Apply diff
     // FIXME only apply the properties that are available in the buffer
-    Trace() << "Applying changed properties: " << changeset;
+    SinkTrace() << "Applying changed properties: " << changeset;
     for (const auto &property : changeset) {
         const auto value = diff->getProperty(property);
         if (value.isValid()) {
@@ -321,7 +322,7 @@ KAsync::Job<qint64> Pipeline::modifiedEntity(void const *command, size_t size)
     }
 
     newAdaptor->resetChangedProperties();
-    for (auto processor : d->processors[bufferType]) {
+    foreach (const auto &processor, d->processors[bufferType]) {
         processor->modifiedEntity(key, Storage::maxRevision(d->transaction) + 1, *current, *newAdaptor, d->transaction);
     }
     //The maxRevision may have changed meanwhile if the entity created sub-entities
@@ -355,7 +356,7 @@ KAsync::Job<qint64> Pipeline::deletedEntity(void const *command, size_t size)
     {
         flatbuffers::Verifier verifyer(reinterpret_cast<const uint8_t *>(command), size);
         if (!Commands::VerifyDeleteEntityBuffer(verifyer)) {
-            Warning() << "invalid buffer, not a delete entity buffer";
+            SinkWarning() << "invalid buffer, not a delete entity buffer";
             return KAsync::error<qint64>(0);
         }
     }
@@ -364,7 +365,7 @@ KAsync::Job<qint64> Pipeline::deletedEntity(void const *command, size_t size)
     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());
-    Log() << "Deleted Entity. Type: " << bufferType << "uid: "<< key << " replayToSource: " << replayToSource;
+    SinkLog() << "Deleted Entity. Type: " << bufferType << "uid: "<< key << " replayToSource: " << replayToSource;
 
     bool found = false;
     bool alreadyRemoved = false;
@@ -381,14 +382,14 @@ KAsync::Job<qint64> Pipeline::deletedEntity(void const *command, size_t size)
                 }
                 return false;
             },
-            [](const Storage::Error &error) { Warning() << "Failed to read old revision from storage: " << error.message; });
+            [](const Storage::Error &error) { SinkWarning() << "Failed to read old revision from storage: " << error.message; });
 
     if (!found) {
-        Warning() << "Failed to find entity " << key;
+        SinkWarning() << "Failed to find entity " << key;
         return KAsync::error<qint64>(0);
     }
     if (alreadyRemoved) {
-        Warning() << "Entity is already removed " << key;
+        SinkWarning() << "Entity is already removed " << key;
         return KAsync::error<qint64>(0);
     }
 
@@ -408,7 +409,7 @@ KAsync::Job<qint64> Pipeline::deletedEntity(void const *command, size_t size)
 
     auto adaptorFactory = Sink::AdaptorFactoryRegistry::instance().getFactory(d->resourceType, bufferType);
     if (!adaptorFactory) {
-        Warning() << "no adaptor factory for type " << bufferType;
+        SinkWarning() << "no adaptor factory for type " << bufferType;
         return KAsync::error<qint64>(0);
     }
 
@@ -418,17 +419,17 @@ KAsync::Job<qint64> Pipeline::deletedEntity(void const *command, size_t size)
             [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";
+                    SinkWarning() << "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; });
+            [this](const Storage::Error &error) { SinkError() << "Failed to find value in pipeline: " << error.message; });
 
     d->storeNewRevision(newRevision, fbb, bufferType, key);
 
-    for (auto processor : d->processors[bufferType]) {
+    foreach (const auto &processor, d->processors[bufferType]) {
         processor->deletedEntity(key, newRevision, *current, d->transaction);
     }
 
@@ -440,13 +441,13 @@ void Pipeline::cleanupRevision(qint64 revision)
     d->revisionChanged = true;
     const auto uid = Storage::getUidFromRevision(d->transaction, revision);
     const auto bufferType = Storage::getTypeFromRevision(d->transaction, revision);
-    Trace() << "Cleaning up revision " << revision << uid << bufferType;
+    SinkTrace() << "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";
+                    SinkWarning() << "Read invalid buffer from disk";
                 } else {
                     const auto metadata = flatbuffers::GetRoot<Metadata>(buffer.metadataBuffer());
                     const qint64 rev = metadata->revision();
@@ -459,7 +460,7 @@ void Pipeline::cleanupRevision(qint64 revision)
 
                 return true;
             },
-            [](const Storage::Error &error) { Warning() << "Error while reading: " << error.message; }, true);
+            [](const Storage::Error &error) { SinkWarning() << "Error while reading: " << error.message; }, true);
     Storage::setCleanedUpRevision(d->transaction, revision);
 }
 
@@ -481,7 +482,6 @@ Preprocessor::Preprocessor() : d(new Preprocessor::Private)
 
 Preprocessor::~Preprocessor()
 {
-    delete d;
 }
 
 void Preprocessor::setup(const QByteArray &resourceType, const QByteArray &resourceInstanceIdentifier, Pipeline *pipeline)