Kill all commands on failing to connect to a resource.

We have to kill pending commands as well, and we have to make sure that
we call open only once the commands are actually enqueued, so we can
kill them in case of failure.

1 files changed, 17 insertions, 16 deletions

diff --git a/common/resourceaccess.cpp b/common/resourceaccess.cpp
index 22d4cdb..29d5a1c 100644
--- a/common/resourceaccess.cpp
+++ b/common/resourceaccess.cpp
@@ -115,6 +115,10 @@ void ResourceAccess::Private::abortPendingOperations()
     for (auto handler : handlers) {
         handler(1, "The resource closed unexpectedly");
     }
+    for (auto queuedCommand : commandQueue) {
+        queuedCommand->callback(1, "The resource closed unexpectedly");
+    }
+    commandQueue.clear();
 }
 
 void ResourceAccess::Private::callCallbacks()
@@ -214,6 +218,7 @@ KAsync::Job<void> ResourceAccess::Private::initializeSocket()
                                 });
                         } else {
                             SinkWarning() << "Failed to start resource";
+                            return KAsync::error(-1, "Failed to start resource");
                         }
                         return KAsync::null();
                     } else {
@@ -256,6 +261,16 @@ void ResourceAccess::registerCallback(uint messageId, const std::function<void(i
     d->resultHandler.insert(messageId, callback);
 }
 
+void ResourceAccess::enqueueCommand(const QSharedPointer<QueuedCommand> &command)
+{
+    d->commandQueue << command;
+    if (isReady()) {
+        processCommandQueue();
+    } else {
+        open();
+    }
+}
+
 KAsync::Job<void> ResourceAccess::sendCommand(int commandId)
 {
     return KAsync::start<void>([this, commandId](KAsync::Future<void> &f) {
@@ -265,10 +280,7 @@ KAsync::Job<void> ResourceAccess::sendCommand(int commandId)
             }
             f.setFinished();
         };
-        d->commandQueue << QSharedPointer<QueuedCommand>::create(commandId, continuation);
-        if (isReady()) {
-            processCommandQueue();
-        }
+        enqueueCommand(QSharedPointer<QueuedCommand>::create(commandId, continuation));
     });
 }
 
@@ -284,11 +296,7 @@ KAsync::Job<void> ResourceAccess::sendCommand(int commandId, flatbuffers::FlatBu
                 f.setFinished();
             }
         };
-
-        d->commandQueue << QSharedPointer<QueuedCommand>::create(commandId, buffer, callback);
-        if (isReady()) {
-            processCommandQueue();
-        }
+        enqueueCommand(QSharedPointer<QueuedCommand>::create(commandId, buffer, callback));
     });
 }
 
@@ -305,7 +313,6 @@ KAsync::Job<void> ResourceAccess::synchronizeResource(const Sink::QueryBase &que
     builder.add_query(q);
     Sink::Commands::FinishSynchronizeBuffer(fbb, builder.Finish());
 
-    open();
     return sendCommand(Commands::SynchronizeCommand, fbb);
 }
 
@@ -318,7 +325,6 @@ KAsync::Job<void> ResourceAccess::sendCreateCommand(const QByteArray &uid, const
     auto delta = Sink::EntityBuffer::appendAsVector(fbb, buffer.constData(), buffer.size());
     auto location = Sink::Commands::CreateCreateEntity(fbb, entityId, type, delta);
     Sink::Commands::FinishCreateEntityBuffer(fbb, location);
-    open();
     return sendCommand(Sink::Commands::CreateEntityCommand, fbb);
 }
 
@@ -334,7 +340,6 @@ ResourceAccess::sendModifyCommand(const QByteArray &uid, qint64 revision, const
     auto resource = newResource.isEmpty() ? 0 : fbb.CreateString(newResource.constData());
     auto location = Sink::Commands::CreateModifyEntity(fbb, revision, entityId, deletions, type, delta, true, modifiedProperties, resource, remove);
     Sink::Commands::FinishModifyEntityBuffer(fbb, location);
-    open();
     return sendCommand(Sink::Commands::ModifyEntityCommand, fbb);
 }
 
@@ -345,7 +350,6 @@ KAsync::Job<void> ResourceAccess::sendDeleteCommand(const QByteArray &uid, qint6
     auto type = fbb.CreateString(resourceBufferType.constData());
     auto location = Sink::Commands::CreateDeleteEntity(fbb, revision, entityId, type);
     Sink::Commands::FinishDeleteEntityBuffer(fbb, location);
-    open();
     return sendCommand(Sink::Commands::DeleteEntityCommand, fbb);
 }
 
@@ -354,7 +358,6 @@ KAsync::Job<void> ResourceAccess::sendRevisionReplayedCommand(qint64 revision)
     flatbuffers::FlatBufferBuilder fbb;
     auto location = Sink::Commands::CreateRevisionReplayed(fbb, revision);
     Sink::Commands::FinishRevisionReplayedBuffer(fbb, location);
-    open();
     return sendCommand(Sink::Commands::RevisionReplayedCommand, fbb);
 }
 
@@ -374,7 +377,6 @@ ResourceAccess::sendInspectionCommand(int inspectionType, const QByteArray &insp
     auto expected = fbb.CreateString(array.toStdString());
     auto location = Sink::Commands::CreateInspection(fbb, id, inspectionType, entity, domain, prop, expected);
     Sink::Commands::FinishInspectionBuffer(fbb, location);
-    open();
     return sendCommand(Sink::Commands::InspectionCommand, fbb);
 }
 
@@ -384,7 +386,6 @@ KAsync::Job<void> ResourceAccess::sendFlushCommand(int flushType, const QByteArr
     auto id = fbb.CreateString(flushId.toStdString());
     auto location = Sink::Commands::CreateFlush(fbb, id, flushType);
     Sink::Commands::FinishFlushBuffer(fbb, location);
-    open();
     return sendCommand(Sink::Commands::FlushCommand, fbb);
 }