Shutdown notification to achieve a clean shutdown.

Otherwise the client always restarts the resource because of the lost connection.
We currently require this in tests to be able to delete the db, but eventually
we likely want a "disable akonadi" function that shuts resources down,
and keeps clients from restarting them (e.g. via configuration).

2 files changed, 26 insertions, 11 deletions

diff --git a/synchronizer/listener.cpp b/synchronizer/listener.cpp
index 2bc6be0..5165111 100644
--- a/synchronizer/listener.cpp
+++ b/synchronizer/listener.cpp
@@ -30,6 +30,7 @@
 #include "common/handshake_generated.h"
 #include "common/revisionupdate_generated.h"
 #include "common/synchronize_generated.h"
+#include "common/notification_generated.h"
 
 #include <QLocalSocket>
 #include <QTimer>
@@ -67,8 +68,7 @@ Listener::Listener(const QString &resourceName, QObject *parent)
     connect(m_checkConnectionsTimer, &QTimer::timeout, [this]() {
         if (m_connections.isEmpty()) {
             Log() << QString("No connections, shutting down.");
-            m_server->close();
-            emit noClients();
+            quit();
         }
     });
 
@@ -250,10 +250,8 @@ void Listener::processCommand(int commandId, uint messageId, Client &client, uin
             break;
         case Akonadi2::Commands::ShutdownCommand:
             Log() << QString("\tReceived shutdown command from %1").arg(client.name);
-            callback();
-            m_server->close();
-            emit noClients();
-            return;
+            QTimer::singleShot(0, this, &Listener::quit);
+            break;
         default:
             if (commandId > Akonadi2::Commands::CustomCommand) {
                 loadResource();
@@ -268,6 +266,22 @@ void Listener::processCommand(int commandId, uint messageId, Client &client, uin
     callback();
 }
 
+void Listener::quit()
+{
+    //Broadcast shutdown notifications to open clients, so they don't try to restart the resource
+    auto command = Akonadi2::CreateNotification(m_fbb, Akonadi2::NotificationType::NotificationType_Shutdown);
+    Akonadi2::FinishNotificationBuffer(m_fbb, command);
+    for (Client &client : m_connections) {
+        if (client.socket && client.socket->isOpen()) {
+            Akonadi2::Commands::write(client.socket, ++m_messageId, Akonadi2::Commands::NotificationCommand, m_fbb);
+        }
+    }
+    m_fbb.Clear();
+
+    m_server->close();
+    emit noClients();
+}
+
 bool Listener::processClientBuffer(Client &client)
 {
     static const int headerSize = Akonadi2::Commands::headerSize();
@@ -314,15 +328,15 @@ void Listener::sendCurrentRevision(Client &client)
     m_fbb.Clear();
 }
 
-void Listener::sendCommandCompleted(Client &client, uint messageId)
+void Listener::sendCommandCompleted(QLocalSocket *socket, uint messageId)
 {
-    if (!client.socket || !client.socket->isValid()) {
+    if (!socket || !socket->isValid()) {
         return;
     }
 
     auto command = Akonadi2::CreateCommandCompletion(m_fbb, messageId);
     Akonadi2::FinishCommandCompletionBuffer(m_fbb, command);
-    Akonadi2::Commands::write(client.socket, ++m_messageId, Akonadi2::Commands::CommandCompletion, m_fbb);
+    Akonadi2::Commands::write(socket, ++m_messageId, Akonadi2::Commands::CommandCompletion, m_fbb);
     m_fbb.Clear();
 }
 
diff --git a/synchronizer/listener.h b/synchronizer/listener.h
index ee73766..8dad3a4 100644
--- a/synchronizer/listener.h
+++ b/synchronizer/listener.h
@@ -49,7 +49,7 @@ public:
     }
 
     QString name;
-    QLocalSocket *socket;
+    QPointer<QLocalSocket> socket;
     QByteArray commandBuffer;
 };
 
@@ -74,12 +74,13 @@ private Q_SLOTS:
     void readFromSocket();
     void processClientBuffers();
     void refreshRevision();
+    void quit();
 
 private:
     void processCommand(int commandId, uint messageId, Client &client, uint size, const std::function<void()> &callback);
     bool processClientBuffer(Client &client);
     void sendCurrentRevision(Client &client);
-    void sendCommandCompleted(Client &client, uint messageId);
+    void sendCommandCompleted(QLocalSocket *socket, uint messageId);
     void updateClientsWithRevision();
     void loadResource();