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).

5 files changed, 50 insertions, 11 deletions

diff --git a/common/commands.h b/common/commands.h
index 26729dc..0007ffc 100644
--- a/common/commands.h
+++ b/common/commands.h
@@ -43,6 +43,7 @@ enum CommandIds {
     CreateEntityCommand,
     SearchSourceCommand, // need a buffer definition for this, but relies on Query API
     ShutdownCommand,
+    NotificationCommand,
     CustomCommand = 0xffff
 };
 
diff --git a/common/commands/notification.fbs b/common/commands/notification.fbs
new file mode 100644
index 0000000..6684472
--- /dev/null
+++ b/common/commands/notification.fbs
@@ -0,0 +1,9 @@
+namespace Akonadi2;
+
+enum NotificationType : byte { Shutdown = 1, Status, Warning, Progress }
+
+table Notification {
+    type: NotificationType = Status;
+}
+
+root_type Notification;
diff --git a/common/resourceaccess.cpp b/common/resourceaccess.cpp
index 59cbece..20e1e8c 100644
--- a/common/resourceaccess.cpp
+++ b/common/resourceaccess.cpp
@@ -25,6 +25,7 @@
 #include "common/handshake_generated.h"
 #include "common/revisionupdate_generated.h"
 #include "common/synchronize_generated.h"
+#include "common/notification_generated.h"
 #include "log.h"
 
 #include <QCoreApplication>
@@ -325,6 +326,19 @@ bool ResourceAccess::processMessageBuffer()
             QMetaObject::invokeMethod(this, "callCallbacks", Qt::QueuedConnection, QGenericReturnArgument(), Q_ARG(int, buffer->id()));
             break;
         }
+        case Commands::NotificationCommand: {
+            auto buffer = GetNotification(d->partialMessageBuffer.constData() + headerSize);
+            switch (buffer->type()) {
+                case Akonadi2::NotificationType::NotificationType_Shutdown:
+                    Log() << "Received shutdown notification.";
+                    close();
+                    break;
+                default:
+                    Warning() << "Received unknown notification: " << buffer->type();
+                    break;
+            }
+            break;
+        }
         default:
             break;
     }
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();