8 files changed, 63 insertions, 45 deletions

diff --git a/common/clientapi.cpp b/common/clientapi.cpp
index 10115da..6f0b421 100644
--- a/common/clientapi.cpp
+++ b/common/clientapi.cpp
@@ -36,10 +36,7 @@ void Store::shutdown(const QString &identifier)
     Akonadi2::ResourceAccess resourceAccess(identifier);
     //FIXME this starts the resource, just to shut it down again if it's not running in the first place.
     resourceAccess.open();
-    resourceAccess.sendCommand(Akonadi2::Commands::ShutdownCommand).then<void>([](Async::Future<void> &f){
-        //TODO wait for disconnect
-        f.setFinished();
-    }).exec().waitForFinished();
+    resourceAccess.sendCommand(Akonadi2::Commands::ShutdownCommand).exec().waitForFinished();
 }
 
 } // namespace Akonadi2
diff --git a/common/clientapi.h b/common/clientapi.h
index aa3aab8..63305ab 100644
--- a/common/clientapi.h
+++ b/common/clientapi.h
@@ -425,13 +425,22 @@ public:
                 auto facade = FacadeFactory::instance().getFacade<DomainType>(resource);
                 //We have to bind an instance to the function callback. Since we use a shared pointer this keeps the result provider instance (and thus also the emitter) alive.
                 std::function<void(const typename DomainType::Ptr &)> addCallback = std::bind(&ResultProvider<typename DomainType::Ptr>::add, resultSet, std::placeholders::_1);
-                //We copy the facade pointer to keep it alive
-                job = job.then(facade->load(query, addCallback));
+
+                // TODO The following is a necessary hack to keep the facade alive.
+                // Otherwise this would reduce to:
+                // job = job.then(facade->load(query, addCallback));
+                // We somehow have to guarantee that the facade remains valid for the duration of the job
+                job = job.then<void>([facade, query, addCallback](Async::Future<void> &future) {
+                    Async::Job<void> j = facade->load(query, addCallback);
+                    j.then<void>([&future, facade](Async::Future<void> &f) {
+                        future.setFinished();
+                        f.setFinished();
+                    }).exec();
+                });
             }
-            job.then<void>([/* eventloop,  */resultSet](Async::Future<void> &future) {
+            job.then<void>([resultSet]() {
                 qDebug() << "Query complete";
                 resultSet->complete();
-                future.setFinished();
             }).exec().waitForFinished(); //We use the eventloop provided by waitForFinished to keep the thread alive until all is done
         });
         return resultSet->emitter();
diff --git a/common/pipeline.cpp b/common/pipeline.cpp
index f321cf5..ed40699 100644
--- a/common/pipeline.cpp
+++ b/common/pipeline.cpp
@@ -30,6 +30,7 @@
 #include "createentity_generated.h"
 #include "entitybuffer.h"
 #include "async/src/async.h"
+#include "log.h"
 
 namespace Akonadi2
 {
@@ -95,7 +96,7 @@ void Pipeline::null()
 
 Async::Job<void> Pipeline::newEntity(void const *command, size_t size)
 {
-    qDebug() << "Pipeline: New Entity";
+    Log() << "Pipeline: New Entity";
 
     //TODO toRFC4122 would probably be more efficient, but results in non-printable keys.
     const auto key = QUuid::createUuid().toString().toUtf8();
@@ -136,7 +137,7 @@ Async::Job<void> Pipeline::newEntity(void const *command, size_t size)
 
     storage().write(key.data(), key.size(), fbb.GetBufferPointer(), fbb.GetSize());
     storage().setMaxRevision(newRevision);
-    qDebug() << "Pipeline: wrote entity: "<< newRevision;
+    Log() << "Pipeline: wrote entity: "<< newRevision;
 
     return Async::start<void>([this, key, entityType](Async::Future<void> &future) {
         PipelineState state(this, NewPipeline, key, d->newPipeline[entityType], [&future]() {
diff --git a/common/resourceaccess.cpp b/common/resourceaccess.cpp
index 22d7d97..7320e50 100644
--- a/common/resourceaccess.cpp
+++ b/common/resourceaccess.cpp
@@ -41,23 +41,14 @@ class QueuedCommand
 public:
     QueuedCommand(int commandId, const std::function<void(int, const QString &)> &callback)
         : commandId(commandId),
-          bufferSize(0),
-          buffer(0),
           callback(callback)
     {}
 
-    QueuedCommand(int commandId, flatbuffers::FlatBufferBuilder &fbb, const std::function<void(int, const QString &)> &callback)
+    QueuedCommand(int commandId, const QByteArray &b, const std::function<void(int, const QString &)> &callback)
         : commandId(commandId),
-          bufferSize(fbb.GetSize()),
-          buffer(new char[bufferSize]),
+          buffer(b),
           callback(callback)
     {
-        memcpy(buffer, fbb.GetBufferPointer(), bufferSize);
-    }
-
-    ~QueuedCommand()
-    {
-        delete[] buffer;
     }
 
 private:
@@ -66,8 +57,7 @@ private:
 
 public:
     const int commandId;
-    const uint bufferSize;
-    char *buffer;
+    QByteArray buffer;
     std::function<void(int, const QString &)> callback;
 };
 
@@ -159,7 +149,9 @@ Async::Job<void> ResourceAccess::sendCommand(int commandId)
 
 Async::Job<void>  ResourceAccess::sendCommand(int commandId, flatbuffers::FlatBufferBuilder &fbb)
 {
-    return Async::start<void>([commandId, &fbb, this](Async::Future<void> &f) {
+    //The flatbuffer is transient, but we want to store it until the job is executed
+    QByteArray buffer(reinterpret_cast<const char*>(fbb.GetBufferPointer()), fbb.GetSize());
+    return Async::start<void>([commandId, buffer, this](Async::Future<void> &f) {
         auto callback = [&f](int error, const QString &errorMessage) {
             if (error) {
                 f.setError(error, errorMessage);
@@ -169,12 +161,13 @@ Async::Job<void>  ResourceAccess::sendCommand(int commandId, flatbuffers::FlatBu
         };
 
         if (isReady()) {
+            //TODO: We probably always want to queue the command, so we can resend it in case something goes wrong
             d->messageId++;
             log(QString("Sending command %1 with messageId %2").arg(commandId).arg(d->messageId));
             registerCallback(d->messageId, callback);
-            Commands::write(d->socket, d->messageId, commandId, fbb);
+            Commands::write(d->socket, d->messageId, commandId, buffer.constData(), buffer.size());
         } else {
-            d->commandQueue << new QueuedCommand(commandId, fbb, callback);
+            d->commandQueue << new QueuedCommand(commandId, buffer, callback);
         }
     });
 }
@@ -237,7 +230,7 @@ void ResourceAccess::connected()
         if (command->callback) {
             registerCallback(d->messageId, command->callback);
         }
-        Commands::write(d->socket, d->messageId, command->commandId, command->buffer, command->bufferSize);
+        Commands::write(d->socket, d->messageId, command->commandId, command->buffer.constData(), command->buffer.size());
         delete command;
     }
     d->commandQueue.clear();
diff --git a/dummyresource/domainadaptor.cpp b/dummyresource/domainadaptor.cpp
index d902052..2182f9a 100644
--- a/dummyresource/domainadaptor.cpp
+++ b/dummyresource/domainadaptor.cpp
@@ -100,13 +100,13 @@ QSharedPointer<Akonadi2::Domain::BufferAdaptor> DummyEventAdaptorFactory::create
         }
     }
 
-    Akonadi2::Metadata const *metadataBuffer = 0;
-    if (auto metadataData = entity.metadata()) {
-        flatbuffers::Verifier verifyer(metadataData->Data(), metadataData->size());
-        if (Akonadi2::VerifyMetadataBuffer(verifyer)) {
-            metadataBuffer = Akonadi2::GetMetadata(metadataData->Data());
-        }
-    }
+    // Akonadi2::Metadata const *metadataBuffer = 0;
+    // if (auto metadataData = entity.metadata()) {
+    //     flatbuffers::Verifier verifyer(metadataData->Data(), metadataData->size());
+    //     if (Akonadi2::VerifyMetadataBuffer(verifyer)) {
+    //         metadataBuffer = Akonadi2::GetMetadata(metadataData->Data());
+    //     }
+    // }
 
     Akonadi2::Domain::Buffer::Event const *localBuffer = 0;
     if (auto localData = entity.local()) {
diff --git a/dummyresource/resourcefactory.cpp b/dummyresource/resourcefactory.cpp
index 10c8eaf..6e06250 100644
--- a/dummyresource/resourcefactory.cpp
+++ b/dummyresource/resourcefactory.cpp
@@ -122,6 +122,17 @@ public:
 signals:
     void error(int errorCode, const QString &errorMessage);
 
+private:
+    bool messagesToProcessAvailable()
+    {
+        for (auto queue : mCommandQueues) {
+            if (!queue->isEmpty()) {
+                return true;
+            }
+        }
+        return false;
+    }
+
 private slots:
     void process()
     {
@@ -131,11 +142,15 @@ private slots:
         mProcessingLock = true;
         auto job = processPipeline().then<void>([this]() {
             mProcessingLock = false;
+            if (messagesToProcessAvailable()) {
+                process();
+            }
         }).exec();
     }
 
     Async::Job<void> processQueuedCommand(const Akonadi2::QueuedCommand *queuedCommand)
     {
+        Log() << "Processing command: " << queuedCommand->commandId();
         //Throw command into appropriate pipeline
         switch (queuedCommand->commandId()) {
             case Akonadi2::Commands::DeleteEntityCommand:
@@ -170,7 +185,7 @@ private slots:
 
                         flatbuffers::Verifier verifyer(reinterpret_cast<const uint8_t *>(ptr), size);
                         if (!Akonadi2::VerifyQueuedCommandBuffer(verifyer)) {
-                            qWarning() << "invalid buffer";
+                            Warning() << "invalid buffer";
                             callback(false);
                             return;
                         }
@@ -185,13 +200,13 @@ private slots:
                                 callback(true);
                             },
                             [callback](int errorCode, QString errorMessage) {
-                                Warning() << errorMessage;
+                                Warning() << "Error while processing queue command: " << errorMessage;
                                 callback(false);
                             }
                         ).exec();
                     },
                     [&future](const MessageQueue::Error &error) {
-                        Warning() << error.message;
+                        Warning() << "Error while getting message from messagequeue: " << error.message;
                         future.setValue(false);
                         future.setFinished();
                     }
@@ -209,6 +224,7 @@ private slots:
             [it, this](Async::Future<void> &future) {
                 auto queue = it->next();
                 processQueue(queue).then<void>([&future]() {
+                    Trace() << "Queue processed";
                     future.setFinished();
                 }).exec();
             }
diff --git a/synchronizer/listener.cpp b/synchronizer/listener.cpp
index 5165111..2e1e918 100644
--- a/synchronizer/listener.cpp
+++ b/synchronizer/listener.cpp
@@ -254,11 +254,13 @@ void Listener::processCommand(int commandId, uint messageId, Client &client, uin
             break;
         default:
             if (commandId > Akonadi2::Commands::CustomCommand) {
+                Log() << QString("\tReceived custom command from %1: ").arg(client.name) << commandId;
                 loadResource();
                 if (m_resource) {
                     m_resource->processCommand(commandId, client.commandBuffer, size, m_pipeline);
                 }
             } else {
+                Warning() << QString("\tReceived invalid command from %1: ").arg(client.name) << commandId;
                 //TODO: handle error: we don't know wtf this command is
             }
             break;
@@ -376,7 +378,7 @@ void Listener::loadResource()
         Log() << QString("\tFacades: %1").arg(Akonadi2::FacadeFactory::instance().getFacade<Akonadi2::Domain::Event>(m_resourceName)->type());
         m_resource->configurePipeline(m_pipeline);
     } else {
-        Warning() << QString("Failed to load resource %1").arg(m_resourceName);
+        Error() << QString("Failed to load resource %1").arg(m_resourceName);
     }
     //TODO: on failure ... what?
     //Enter broken state?
diff --git a/tests/domainadaptortest.cpp b/tests/domainadaptortest.cpp
index d1a9d26..cedbf94 100644
--- a/tests/domainadaptortest.cpp
+++ b/tests/domainadaptortest.cpp
@@ -74,13 +74,13 @@ public:
             }
         }
 
-        Akonadi2::Metadata const *metadataBuffer = 0;
-        if (auto metadataData = entity.metadata()) {
-            flatbuffers::Verifier verifyer(metadataData->Data(), metadataData->size());
-            if (Akonadi2::VerifyMetadataBuffer(verifyer)) {
-                metadataBuffer = Akonadi2::GetMetadata(metadataData);
-            }
-        }
+        // Akonadi2::Metadata const *metadataBuffer = 0;
+        // if (auto metadataData = entity.metadata()) {
+        //     flatbuffers::Verifier verifyer(metadataData->Data(), metadataData->size());
+        //     if (Akonadi2::VerifyMetadataBuffer(verifyer)) {
+        //         metadataBuffer = Akonadi2::GetMetadata(metadataData);
+        //     }
+        // }
 
         Akonadi2::Domain::Buffer::Event const *localBuffer = 0;
         if (auto localData = entity.local()) {