send revision updates from resource to client

this includes an initial revision message on connect

7 files changed, 136 insertions, 12 deletions

diff --git a/client/resourceaccess.cpp b/client/resourceaccess.cpp
index 4042219..e73c773 100644
--- a/client/resourceaccess.cpp
+++ b/client/resourceaccess.cpp
@@ -3,6 +3,7 @@
 #include "common/console.h"
 #include "common/commands.h"
 #include "common/handshake_generated.h"
+#include "common/revisionupdate_generated.h"
 
 #include <QDebug>
 #include <QProcess>
@@ -26,6 +27,9 @@ ResourceAccess::ResourceAccess(const QString &resourceName, QObject *parent)
             this, &ResourceAccess::disconnected);
     connect(m_socket, SIGNAL(error(QLocalSocket::LocalSocketError)),
             this, SLOT(connectionError(QLocalSocket::LocalSocketError)));
+    connect(m_socket, &QIODevice::readyRead,
+            this, &ResourceAccess::readResourceMessage);
+
 }
 
 ResourceAccess::~ResourceAccess()
@@ -109,3 +113,44 @@ void ResourceAccess::connectionError(QLocalSocket::LocalSocketError error)
     }
 }
 
+void ResourceAccess::readResourceMessage()
+{
+    if (!m_socket->isValid()) {
+        return;
+    }
+
+    m_partialMessageBuffer += m_socket->readAll();
+
+    // should be scheduled rather than processed all at once
+    while (processMessageBuffer()) {}
+}
+
+bool ResourceAccess::processMessageBuffer()
+{
+    static const int headerSize = (sizeof(int) * 2);
+    Console::main()->log(QString("processing %1").arg(m_partialMessageBuffer.size()));
+    if (m_partialMessageBuffer.size() < headerSize) {
+        return false;
+    }
+
+    const int commandId = *(int*)m_partialMessageBuffer.constData();
+    const int size = *(int*)(m_partialMessageBuffer.constData() + sizeof(int));
+
+    if (size > m_partialMessageBuffer.size() - headerSize) {
+        return false;
+    }
+
+    switch (commandId) {
+        case Commands::RevisionUpdateCommand: {
+            auto buffer = Toynadi::GetRevisionUpdate(m_partialMessageBuffer.constData() + headerSize);
+            Console::main()->log(QString("    Revision updated to: %1").arg(buffer->revision()));
+            emit revisionChanged(buffer->revision());
+            break;
+        }
+        default:
+            break;
+    }
+
+    m_partialMessageBuffer.remove(0, headerSize + size);
+    return m_partialMessageBuffer.size() >= headerSize;
+}
diff --git a/client/resourceaccess.h b/client/resourceaccess.h
index 53b46c5..09df614 100644
--- a/client/resourceaccess.h
+++ b/client/resourceaccess.h
@@ -21,15 +21,19 @@ public Q_SLOTS:
 
 Q_SIGNALS:
     void ready(bool isReady);
+    void revisionChanged(unsigned long long revision);
 
 private Q_SLOTS:
     void connected();
     void disconnected();
     void connectionError(QLocalSocket::LocalSocketError error);
+    void readResourceMessage();
+    bool processMessageBuffer();
 
 private:
     QString m_resourceName;
     QLocalSocket *m_socket;
     QTimer *m_tryOpenTimer;
     bool m_startingProcess;
+    QByteArray m_partialMessageBuffer;
 };
diff --git a/common/CMakeLists.txt b/common/CMakeLists.txt
index 32e10f6..a0d3ac9 100644
--- a/common/CMakeLists.txt
+++ b/common/CMakeLists.txt
@@ -1,5 +1,6 @@
 project(toynadicommon)
-generate_flatbuffers(commands/handshake)
+generate_flatbuffers(commands/handshake
+                     commands/revisionupdate)
 
 set(command_SRCS
    console.cpp
diff --git a/common/commands.h b/common/commands.h
index 74174ee..68a7c1e 100644
--- a/common/commands.h
+++ b/common/commands.h
@@ -6,6 +6,7 @@ namespace Commands
 enum CommandIds {
     UnknownCommand = 0,
     HandshakeCommand,
+    RevisionUpdateCommand,
     CustomCommand = 0xffff
 };
 
diff --git a/common/commands/revisionupdate.fbs b/common/commands/revisionupdate.fbs
new file mode 100644
index 0000000..d95f1b5
--- /dev/null
+++ b/common/commands/revisionupdate.fbs
@@ -0,0 +1,7 @@
+namespace Toynadi;
+
+table RevisionUpdate {
+    revision: ulong;
+}
+
+root_type RevisionUpdate;
\ No newline at end of file
diff --git a/resource/listener.cpp b/resource/listener.cpp
index 8a571e9..5dd8cc3 100644
--- a/resource/listener.cpp
+++ b/resource/listener.cpp
@@ -3,13 +3,15 @@
 #include "common/console.h"
 #include "common/commands.h"
 #include "common/handshake_generated.h"
+#include "common/revisionupdate_generated.h"
 
 #include <QLocalSocket>
 #include <QTimer>
 
 Listener::Listener(const QString &resource, QObject *parent)
     : QObject(parent),
-      m_server(new QLocalServer(this))
+      m_server(new QLocalServer(this)),
+      m_revision(0)
 {
     connect(m_server, &QLocalServer::newConnection,
              this, &Listener::acceptConnection);
@@ -34,6 +36,19 @@ Listener::~Listener()
 {
 }
 
+void Listener::setRevision(unsigned long long revision)
+{
+    if (m_revision != revision) {
+        m_revision = revision;
+        updateClientsWithRevision();
+    }
+}
+
+unsigned long long Listener::revision() const
+{
+    return m_revision;
+}
+
 void Listener::closeAllConnections()
 {
     //TODO: close all client connectionsin m_connections
@@ -73,7 +88,7 @@ void Listener::clientDropped()
     }
 
     Console::main()->log("Dropping connection...");
-    QMutableListIterator<Client> it(m_connections);
+    QMutableVectorIterator<Client> it(m_connections);
     while (it.hasNext()) {
         const Client &client = it.next();
         if (client.socket == socket) {
@@ -102,24 +117,25 @@ void Listener::readFromSocket()
     }
 
     Console::main()->log("Reading from socket...");
-    QMutableListIterator<Client> it(m_connections);
-    while (it.hasNext()) {
-        Client &client = it.next();
+    for (Client &client: m_connections) {
         if (client.socket == socket) {
             Console::main()->log(QString("    Client: %1").arg(client.name));
             client.commandBuffer += socket->readAll();
-            processClientBuffer(client);
+            // FIXME: schedule these rather than process them all at once
+            //        right now this can lead to starvation of clients due to
+            //        one overly active client
+            while (processClientBuffer(client)) {}
             break;
         }
     }
 }
 
-void Listener::processClientBuffer(Client &client)
+bool Listener::processClientBuffer(Client &client)
 {
     static const int headerSize = (sizeof(int) * 2);
     Console::main()->log(QString("processing %1").arg(client.commandBuffer.size()));
     if (client.commandBuffer.size() < headerSize) {
-        return;
+        return false;
     }
 
     int commandId, size;
@@ -134,12 +150,50 @@ void Listener::processClientBuffer(Client &client)
             case Commands::HandshakeCommand: {
                 auto buffer = Toynadi::GetHandshake(data.constData());
                 Console::main()->log(QString("    Handshake from %1").arg(buffer->name()->c_str()));
-                //TODO: reply?
+                sendCurrentRevision(client);
                 break;
             }
             default:
                 // client.hasSentCommand = true;
                 break;
         }
+
+        return client.commandBuffer.size() >= headerSize;
+    } else {
+        return false;
+    }
+}
+
+void Listener::sendCurrentRevision(Client &client)
+{
+    if (!client.socket || !client.socket->isValid()) {
+        return;
+    }
+
+    flatbuffers::FlatBufferBuilder fbb;
+    auto command = Toynadi::CreateRevisionUpdate(fbb, m_revision);
+    Toynadi::FinishRevisionUpdateBuffer(fbb, command);
+    const int commandId = Commands::RevisionUpdateCommand;
+    const int dataSize = fbb.GetSize();
+    client.socket->write((const char*)&commandId, sizeof(int));
+    client.socket->write((const char*)&dataSize, sizeof(int));
+    client.socket->write((const char*)fbb.GetBufferPointer(), dataSize);
+}
+
+void Listener::updateClientsWithRevision()
+{
+    flatbuffers::FlatBufferBuilder fbb;
+    auto command = Toynadi::CreateRevisionUpdate(fbb, m_revision);
+    Toynadi::FinishRevisionUpdateBuffer(fbb, command);
+    const int commandId = Commands::RevisionUpdateCommand;
+    const int dataSize = fbb.GetSize();
+
+    for (const Client &client: m_connections) {
+        if (!client.socket || !client.socket->isValid()) {
+            continue;
+        }
+        client.socket->write((const char*)&commandId, sizeof(int));
+        client.socket->write((const char*)&dataSize, sizeof(int));
+        client.socket->write((const char*)fbb.GetBufferPointer(), dataSize);
     }
 }
diff --git a/resource/listener.h b/resource/listener.h
index bfafe5f..0d8f388 100644
--- a/resource/listener.h
+++ b/resource/listener.h
@@ -8,6 +8,12 @@
 class Client
 {
 public:
+    Client()
+        : socket(nullptr),
+          hasSentCommand(false)
+    {
+    }
+
     Client(const QString &n, QLocalSocket *s)
         : name(n),
           socket(s),
@@ -29,6 +35,9 @@ public:
     Listener(const QString &resourceName, QObject *parent = 0);
     ~Listener();
 
+    void setRevision(unsigned long long revision);
+    unsigned long long revision() const;
+
 Q_SIGNALS:
     void noClients();
 
@@ -42,7 +51,10 @@ private Q_SLOTS:
     void readFromSocket();
 
 private:
-    void processClientBuffer(Client &client);
+    bool processClientBuffer(Client &client);
+    void sendCurrentRevision(Client &client);
+    void updateClientsWithRevision();
     QLocalServer *m_server;
-    QList<Client> m_connections;
+    QVector<Client> m_connections;
+    unsigned long long m_revision;
 };
\ No newline at end of file