From aed0d52dbdcecb29b38dcc24fb00d433ff6193e9 Mon Sep 17 00:00:00 2001
From: Sander Speetjens <sander.speetjens@gmail.com>
Date: Thu, 22 Aug 2024 11:19:31 +0200
Subject: [PATCH] websocket

split broadcast out of message function
---
 tools/main.cpp      |  2 ++
 tools/websocket.cpp | 64 ++++++++++++++++++++++++++++-----------------
 tools/websocket.hpp |  1 +
 3 files changed, 43 insertions(+), 24 deletions(-)

diff --git a/tools/main.cpp b/tools/main.cpp
index e1fb76c..fa94b85 100644
--- a/tools/main.cpp
+++ b/tools/main.cpp
@@ -38,9 +38,11 @@ int main(void)
     g_web_socket = &web_socket;
 
     std::thread message_processor_thread(std::bind(&Websocket::process_messages, &web_socket));
+    std::thread broadcast_processor_thred(std::bind(&Websocket::process_broadcast, &web_socket));
     
     // Run the asio loop with the main thread
     web_socket.run();
 
     message_processor_thread.join();
+    broadcast_processor_thred.join();
 }
\ No newline at end of file
diff --git a/tools/websocket.cpp b/tools/websocket.cpp
index 2e54e7a..fa02813 100644
--- a/tools/websocket.cpp
+++ b/tools/websocket.cpp
@@ -1,7 +1,8 @@
 #include "websocket.hpp"
 
-
-Websocket::Websocket(uint16_t port, std::function<std::string(std::string)> on_message, std::function<std::string()> on_update)
+Websocket::Websocket(uint16_t port,
+                     std::function<std::string(std::string)> on_message,
+                     std::function<std::string()> on_update)
     : m_on_message(on_message), m_on_update(on_update), m_port(port)
 {
     // Set logging settings
@@ -27,7 +28,7 @@ void Websocket::run()
     {
         // Start the ASIO io_service run loop
         m_server.run();
-    } catch (websocketpp::exception const & e) {
+    } catch (websocketpp::exception const& e) {
         std::cout << e.what() << std::endl;
     } catch (...) {
         std::cout << "other exception" << std::endl;
@@ -44,8 +45,8 @@ void Websocket::on_open(websocketpp::connection_hdl hdl)
 {
     {
         std::lock_guard<std::mutex> guard(m_action_lock);
-        //std::cout << "on_open" << std::endl;
-        m_actions.push(action(SUBSCRIBE,hdl));
+        // std::cout << "on_open" << std::endl;
+        m_actions.push(action(SUBSCRIBE, hdl));
     }
     m_action_cond.notify_one();
 }
@@ -54,27 +55,31 @@ void Websocket::on_close(websocketpp::connection_hdl hdl)
 {
     {
         std::lock_guard<std::mutex> guard(m_action_lock);
-        //std::cout << "on_close" << std::endl;
-        m_actions.push(action(UNSUBSCRIBE,hdl));
+        // std::cout << "on_close" << std::endl;
+        m_actions.push(action(UNSUBSCRIBE, hdl));
     }
     m_action_cond.notify_one();
 }
 
-void Websocket::on_message(websocketpp::connection_hdl hdl, Server::message_ptr msg) {
+void Websocket::on_message(websocketpp::connection_hdl hdl, Server::message_ptr msg)
+{
     // queue message up for sending by processing thread
     {
         std::lock_guard<std::mutex> guard(m_action_lock);
-        //std::cout << "on_message" << std::endl;
-        m_actions.push(action(MESSAGE,hdl,msg));
+        // std::cout << "on_message" << std::endl;
+        m_actions.push(action(MESSAGE, hdl, msg));
     }
     m_action_cond.notify_one();
 }
 
-void Websocket::process_messages() {
-    while(1) {
+void Websocket::process_messages()
+{
+    while (1)
+    {
         std::unique_lock<std::mutex> lock(m_action_lock);
 
-        while(m_actions.empty()) {
+        while (m_actions.empty())
+        {
             m_action_cond.wait(lock);
         }
 
@@ -83,38 +88,49 @@ void Websocket::process_messages() {
 
         lock.unlock();
 
-        switch (a.type)
-        {
-        case SUBSCRIBE:
+        switch (a.type) {
+            case SUBSCRIBE:
             {
                 std::lock_guard<std::mutex> guard(m_connection_lock);
                 m_connections.insert(a.hdl);
             }
             break;
-        case UNSUBSCRIBE:
+            case UNSUBSCRIBE:
             {
                 std::lock_guard<std::mutex> guard(m_connection_lock);
                 m_connections.erase(a.hdl);
             }
             break;
-        case MESSAGE:
+            case MESSAGE:
             {
                 std::lock_guard<std::mutex> guard(m_connection_lock);
                 std::string response = m_on_message(a.msg->get_payload());
                 m_server.send(a.hdl, response, websocketpp::frame::opcode::TEXT);
             }
             break;
-        default:
-            // undefined.
-            break;
+            default:
+                // undefined.
+                break;
         }
+    }
+}
 
-        std::string update = m_on_update();
-        if (!update.empty())
+void Websocket::process_broadcast()
+{
+    while (1)
+    {
         {
             std::lock_guard<std::mutex> guard(m_connection_lock);
-            for (auto it = m_connections.begin(); it != m_connections.end(); ++it)
+            if (m_connections.empty())
             {
+                continue;
+            }
+        }
+        std::string update = m_on_update();
+        if (!update.empty() && update != "{}")
+        {
+            std::lock_guard<std::mutex> guard(m_connection_lock);
+            for (auto it = m_connections.begin(); it != m_connections.end(); ++it) {
                 m_server.send(*it, update, websocketpp::frame::opcode::TEXT);
             }
         }
diff --git a/tools/websocket.hpp b/tools/websocket.hpp
index 0109f79..c27a9b8 100644
--- a/tools/websocket.hpp
+++ b/tools/websocket.hpp
@@ -39,6 +39,7 @@ public:
     void run();
     void stop();
     void process_messages();
+    void process_broadcast();
 private:
     void on_open(websocketpp::connection_hdl hdl);
     void on_close(websocketpp::connection_hdl hdl);