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();