6 files changed, 38 insertions, 48 deletions

diff --git a/common/genericresource.cpp b/common/genericresource.cpp
index 2b9e7b2..ed7dd46 100644
--- a/common/genericresource.cpp
+++ b/common/genericresource.cpp
@@ -240,7 +240,7 @@ GenericResource::GenericResource(const QByteArray &resourceType, const QByteArra
       mClientLowerBoundRevision(std::numeric_limits<qint64>::max())
 {
     mPipeline->setResourceType(mResourceType);
-    mProcessor = new CommandProcessor(mPipeline.data(), QList<MessageQueue *>() << &mUserQueue << &mSynchronizerQueue);
+    mProcessor = std::unique_ptr<CommandProcessor>(new CommandProcessor(mPipeline.data(), QList<MessageQueue *>() << &mUserQueue << &mSynchronizerQueue));
     mProcessor->setInspectionCommand([this](void const *command, size_t size) {
         flatbuffers::Verifier verifier((const uint8_t *)command, size);
         if (Sink::Commands::VerifyInspectionBuffer(verifier)) {
@@ -280,7 +280,7 @@ GenericResource::GenericResource(const QByteArray &resourceType, const QByteArra
         return KAsync::error<void>(-1, "Invalid inspection command.");
     });
     {
-        auto ret =QObject::connect(mProcessor, &CommandProcessor::error, [this](int errorCode, const QString &msg) { onProcessorError(errorCode, msg); });
+        auto ret =QObject::connect(mProcessor.get(), &CommandProcessor::error, [this](int errorCode, const QString &msg) { onProcessorError(errorCode, msg); });
         Q_ASSERT(ret);
     }
     {
@@ -296,7 +296,6 @@ GenericResource::GenericResource(const QByteArray &resourceType, const QByteArra
 
 GenericResource::~GenericResource()
 {
-    delete mProcessor;
 }
 
 KAsync::Job<void> GenericResource::inspect(
diff --git a/common/genericresource.h b/common/genericresource.h
index 0878968..2254172 100644
--- a/common/genericresource.h
+++ b/common/genericresource.h
@@ -77,7 +77,7 @@ protected:
     QSharedPointer<Pipeline> mPipeline;
 
 private:
-    CommandProcessor *mProcessor;
+    std::unique_ptr<CommandProcessor> mProcessor;
     QSharedPointer<ChangeReplay> mChangeReplay;
     QSharedPointer<Synchronizer> mSynchronizer;
     int mError;
diff --git a/common/listener.cpp b/common/listener.cpp
index 32c57ac..af8eaa2 100644
--- a/common/listener.cpp
+++ b/common/listener.cpp
@@ -47,11 +47,10 @@ Listener::Listener(const QByteArray &resourceInstanceIdentifier, const QByteArra
       m_server(new QLocalServer(this)),
       m_resourceName(resourceType),
       m_resourceInstanceIdentifier(resourceInstanceIdentifier),
-      m_resource(0),
       m_clientBufferProcessesTimer(new QTimer(this)),
       m_messageId(0)
 {
-    connect(m_server, &QLocalServer::newConnection, this, &Listener::acceptConnection);
+    connect(m_server.get(), &QLocalServer::newConnection, this, &Listener::acceptConnection);
     Trace() << "Trying to open " << m_resourceInstanceIdentifier;
 
     if (!m_server->listen(QString::fromLatin1(m_resourceInstanceIdentifier))) {
@@ -66,10 +65,10 @@ Listener::Listener(const QByteArray &resourceInstanceIdentifier, const QByteArra
         Log() << QString("Listening on %1").arg(m_server->serverName());
     }
 
-    m_checkConnectionsTimer = new QTimer;
+    m_checkConnectionsTimer = std::unique_ptr<QTimer>(new QTimer);
     m_checkConnectionsTimer->setSingleShot(true);
     m_checkConnectionsTimer->setInterval(1000);
-    connect(m_checkConnectionsTimer, &QTimer::timeout, [this]() {
+    connect(m_checkConnectionsTimer.get(), &QTimer::timeout, [this]() {
         if (m_connections.isEmpty()) {
             Log() << QString("No connections, shutting down.");
             quit();
@@ -80,16 +79,12 @@ Listener::Listener(const QByteArray &resourceInstanceIdentifier, const QByteArra
     //      or even just drop down to invoking the method queued? => invoke queued unless we need throttling
     m_clientBufferProcessesTimer->setInterval(0);
     m_clientBufferProcessesTimer->setSingleShot(true);
-    connect(m_clientBufferProcessesTimer, &QTimer::timeout, this, &Listener::processClientBuffers);
+    connect(m_clientBufferProcessesTimer.get(), &QTimer::timeout, this, &Listener::processClientBuffers);
 }
 
 Listener::~Listener()
 {
     closeAllConnections();
-    delete m_resource;
-    delete m_checkConnectionsTimer;
-    delete m_clientBufferProcessesTimer;
-    delete m_server;
 }
 
 void Listener::emergencyAbortAllConnections()
@@ -140,7 +135,7 @@ void Listener::acceptConnection()
 
     // If this is the first client, set the lower limit for revision cleanup
     if (m_connections.size() == 1) {
-        loadResource()->setLowerBoundRevision(0);
+        loadResource().setLowerBoundRevision(0);
     }
 
     if (socket->bytesAvailable()) {
@@ -177,7 +172,7 @@ void Listener::checkConnections()
 {
     // If this was the last client, disengage the lower limit for revision cleanup
     if (m_connections.isEmpty()) {
-        loadResource()->setLowerBoundRevision(std::numeric_limits<qint64>::max());
+        loadResource().setLowerBoundRevision(std::numeric_limits<qint64>::max());
     }
     m_checkConnectionsTimer->start();
 }
@@ -249,10 +244,10 @@ void Listener::processCommand(int commandId, uint messageId, const QByteArray &c
                 timer->start();
                 auto job = KAsync::null<void>();
                 if (buffer->sourceSync()) {
-                    job = loadResource()->synchronizeWithSource();
+                    job = loadResource().synchronizeWithSource();
                 }
                 if (buffer->localSync()) {
-                    job = job.then<void>(loadResource()->processAllMessages());
+                    job = job.then<void>(loadResource().processAllMessages());
                 }
                 job.then<void>([callback, timer]() {
                        Trace() << "Sync took " << Sink::Log::TraceTime(timer->elapsed());
@@ -274,7 +269,7 @@ void Listener::processCommand(int commandId, uint messageId, const QByteArray &c
         case Sink::Commands::ModifyEntityCommand:
         case Sink::Commands::CreateEntityCommand:
             Trace() << "Command id  " << messageId << " of type \"" << Sink::Commands::name(commandId) << "\" from " << client.name;
-            loadResource()->processCommand(commandId, commandBuffer);
+            loadResource().processCommand(commandId, commandBuffer);
             break;
         case Sink::Commands::ShutdownCommand:
             Log() << QString("Received shutdown command from %1").arg(client.name);
@@ -294,20 +289,19 @@ void Listener::processCommand(int commandId, uint messageId, const QByteArray &c
             } else {
                 Warning() << "received invalid command";
             }
-            loadResource()->setLowerBoundRevision(lowerBoundRevision());
+            loadResource().setLowerBoundRevision(lowerBoundRevision());
         } break;
         case Sink::Commands::RemoveFromDiskCommand: {
             Log() << QString("Received a remove from disk command from %1").arg(client.name);
-            delete m_resource;
-            m_resource = nullptr;
-            loadResource()->removeDataFromDisk();
+            m_resource.reset(nullptr);
+            loadResource().removeDataFromDisk();
             m_server->close();
             QTimer::singleShot(0, this, &Listener::quit);
         } break;
         default:
             if (commandId > Sink::Commands::CustomCommand) {
                 Log() << QString("Received custom command from %1: ").arg(client.name) << commandId;
-                loadResource()->processCommand(commandId, commandBuffer);
+                loadResource().processCommand(commandId, commandBuffer);
             } else {
                 success = false;
                 ErrorMsg() << QString("\tReceived invalid command from %1: ").arg(client.name) << commandId;
@@ -437,25 +431,25 @@ void Listener::notify(const Sink::Notification &notification)
     m_fbb.Clear();
 }
 
-Sink::Resource *Listener::loadResource()
+Sink::Resource &Listener::loadResource()
 {
     if (!m_resource) {
         if (Sink::ResourceFactory *resourceFactory = Sink::ResourceFactory::load(m_resourceName)) {
-            m_resource = resourceFactory->createResource(m_resourceInstanceIdentifier);
+            m_resource = std::unique_ptr<Sink::Resource>(resourceFactory->createResource(m_resourceInstanceIdentifier));
             if (!m_resource) {
                 ErrorMsg() << "Failed to instantiate the resource " << m_resourceName;
-                m_resource = new Sink::Resource;
+                m_resource = std::unique_ptr<Sink::Resource>(new Sink::Resource);
             }
             Trace() << QString("Resource factory: %1").arg((qlonglong)resourceFactory);
-            Trace() << QString("\tResource: %1").arg((qlonglong)m_resource);
-            connect(m_resource, &Sink::Resource::revisionUpdated, this, &Listener::refreshRevision);
-            connect(m_resource, &Sink::Resource::notify, this, &Listener::notify);
+            Trace() << QString("\tResource: %1").arg((qlonglong)m_resource.get());
+            connect(m_resource.get(), &Sink::Resource::revisionUpdated, this, &Listener::refreshRevision);
+            connect(m_resource.get(), &Sink::Resource::notify, this, &Listener::notify);
         } else {
             ErrorMsg() << "Failed to load resource " << m_resourceName;
-            m_resource = new Sink::Resource;
+            m_resource = std::unique_ptr<Sink::Resource>(new Sink::Resource);
         }
     }
-    return m_resource;
+    return *m_resource;
 }
 
 #pragma clang diagnostic push
diff --git a/common/listener.h b/common/listener.h
index 5e376c7..67d76e9 100644
--- a/common/listener.h
+++ b/common/listener.h
@@ -81,17 +81,17 @@ private:
     bool processClientBuffer(Client &client);
     void sendCommandCompleted(QLocalSocket *socket, uint messageId, bool success);
     void updateClientsWithRevision(qint64);
-    Sink::Resource *loadResource();
+    Sink::Resource &loadResource();
     void readFromSocket(QLocalSocket *socket);
     qint64 lowerBoundRevision();
 
-    QLocalServer *m_server;
+    std::unique_ptr<QLocalServer> m_server;
     QVector<Client> m_connections;
     flatbuffers::FlatBufferBuilder m_fbb;
     const QByteArray m_resourceName;
     const QByteArray m_resourceInstanceIdentifier;
-    Sink::Resource *m_resource;
-    QTimer *m_clientBufferProcessesTimer;
-    QTimer *m_checkConnectionsTimer;
+    std::unique_ptr<Sink::Resource> m_resource;
+    std::unique_ptr<QTimer> m_clientBufferProcessesTimer;
+    std::unique_ptr<QTimer> m_checkConnectionsTimer;
     int m_messageId;
 };
diff --git a/common/pipeline.cpp b/common/pipeline.cpp
index feceb77..034f913 100644
--- a/common/pipeline.cpp
+++ b/common/pipeline.cpp
@@ -52,7 +52,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;
@@ -80,18 +80,16 @@ Pipeline::Pipeline(const QString &resourceName, QObject *parent) : QObject(paren
 Pipeline::~Pipeline()
 {
     d->transaction = Storage::Transaction();
-    for (const auto &t : d->processors.keys()) {
-        qDeleteAll(d->processors.value(t));
-    }
-    delete d;
 }
 
 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)
@@ -216,7 +214,7 @@ KAsync::Job<qint64> Pipeline::newEntity(void const *command, size_t size)
 
     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
@@ -325,7 +323,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
@@ -432,7 +430,7 @@ KAsync::Job<qint64> Pipeline::deletedEntity(void const *command, size_t size)
 
     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);
     }
 
@@ -485,7 +483,6 @@ Preprocessor::Preprocessor() : d(new Preprocessor::Private)
 
 Preprocessor::~Preprocessor()
 {
-    delete d;
 }
 
 void Preprocessor::setup(const QByteArray &resourceType, const QByteArray &resourceInstanceIdentifier, Pipeline *pipeline)
diff --git a/common/pipeline.h b/common/pipeline.h
index d04d795..ef89cf0 100644
--- a/common/pipeline.h
+++ b/common/pipeline.h
@@ -72,7 +72,7 @@ signals:
 
 private:
     class Private;
-    Private *const d;
+    const std::unique_ptr<Private> d;
 };
 
 class SINK_EXPORT Preprocessor
@@ -103,7 +103,7 @@ protected:
 private:
     friend class Pipeline;
     class Private;
-    Private *const d;
+    const std::unique_ptr<Private> d;
 };
 
 template<typename DomainType>