Using broadcast server example for server

tools/main.cpp

@@ -1,7 +1,6 @@
 #include <iostream>
 #include <signal.h>
 #include <stdlib.h>
-#include <stdio.h>
 #include <unistd.h>
 
 #include "simulators.hpp"
@@ -32,11 +31,15 @@ int main(void)
 
     sigaction(SIGINT, &sigIntHandler, NULL);
 
-    Websocket web_socket(9002, [](std::string message) {
-        return message;
-    });
-    g_web_socket = &web_socket;
-    web_socket.run();
-    
 
+    // The server starts in this thread
+    Websocket web_socket(9002, [](std::string message) { return message; }, []() { return ""; });
+    g_web_socket = &web_socket;
+
+    std::thread message_processor_thread(std::bind(&Websocket::process_messages, &web_socket));
+    
+    // Run the asio loop with the main thread
+    web_socket.run();
+
+    message_processor_thread.join();
 }

tools/websocket.cpp

@@ -1,38 +1,32 @@
 #include "websocket.hpp"
 
 
-Websocket::Websocket(uint16_t port, std::function<std::string(std::string)> on_message)
-    : p_on_message(on_message), p_port(port)
+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)
 {
-        try {
-        // Set logging settings
-        server.set_access_channels(websocketpp::log::alevel::all);
-        server.clear_access_channels(websocketpp::log::alevel::frame_payload);
+    // Set logging settings
+    m_server.set_access_channels(websocketpp::log::alevel::all);
+    m_server.clear_access_channels(websocketpp::log::alevel::frame_payload);
 
-        // Initialize Asio
-        server.init_asio();
+    // Initialize Asio
+    m_server.init_asio();
 
-        // Register our message handler
-        server.set_open_handler(bind(&Websocket::on_open, this, ::_1));
-        server.set_close_handler(bind(&Websocket::on_close, this, ::_1));
-        server.set_message_handler(bind(&Websocket::on_message, this, &server, ::_1, ::_2));
-    } catch (websocketpp::exception const & e) {
-        std::cout << e.what() << std::endl;
-    } catch (...) {
-        std::cout << "other exception" << std::endl;
-    }
+    // Register our message handler
+    m_server.set_open_handler(std::bind(&Websocket::on_open, this, ::_1));
+    m_server.set_close_handler(std::bind(&Websocket::on_close, this, ::_1));
+    m_server.set_message_handler(std::bind(&Websocket::on_message, this, ::_1, ::_2));
 }
 
 void Websocket::run()
 {
+    // listen on specified port
+    m_server.listen(m_port);
+    // Start the server accept loop
+    m_server.start_accept();
     try
     {
-        server.listen(p_port);
-        // Start the server accept loop
-        server.start_accept();
-
         // Start the ASIO io_service run loop
-        server.run();
+        m_server.run();
     } catch (websocketpp::exception const & e) {
         std::cout << e.what() << std::endl;
     } catch (...) {
@@ -42,19 +36,87 @@ void Websocket::run()
 
 void Websocket::stop()
 {
-    server.stop_listening();
-    server.stop();
+    m_server.stop_listening();
+    m_server.stop();
 }
 
-void Websocket::on_message(Server* s, websocketpp::connection_hdl hdl, Server::message_ptr msg) {
-    std::cout << "on_message called with hdl: " << hdl.lock().get()
-              << " and message: " << msg->get_payload()
-              << std::endl;
-    std::string response = p_on_message(msg->get_payload());
-    try {
-        s->send(hdl, response, msg->get_opcode());
-    } catch (websocketpp::exception const & e) {
-        std::cout << "Response failed because: "
-                  << "(" << e.what() << ")" << std::endl;
+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));
+    }
+    m_action_cond.notify_one();
+}
+
+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));
+    }
+    m_action_cond.notify_one();
+}
+
+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));
+    }
+    m_action_cond.notify_one();
+}
+
+void Websocket::process_messages() {
+    while(1) {
+        std::unique_lock<std::mutex> lock(m_action_lock);
+
+        while(m_actions.empty()) {
+            m_action_cond.wait(lock);
+        }
+
+        action a = m_actions.front();
+        m_actions.pop();
+
+        lock.unlock();
+
+        switch (a.type)
+        {
+        case SUBSCRIBE:
+            {
+                std::lock_guard<std::mutex> guard(m_connection_lock);
+                m_connections.insert(a.hdl);
+            }
+            break;
+        case UNSUBSCRIBE:
+            {
+                std::lock_guard<std::mutex> guard(m_connection_lock);
+                m_connections.erase(a.hdl);
+            }
+            break;
+        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;
+        }
+
+        std::string update = m_on_update();
+        if (!update.empty())
+        {
+            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);
+            }
+        }
     }
 }

tools/websocket.hpp

@@ -1,6 +1,7 @@
 #pragma once
 
 #include <functional>
+#include <set>
 
 #include <websocketpp/config/asio_no_tls.hpp>
 #include <websocketpp/server.hpp>
@@ -9,17 +10,50 @@ typedef websocketpp::server<websocketpp::config::asio> Server;
 
 using websocketpp::lib::placeholders::_1;
 using websocketpp::lib::placeholders::_2;
-using websocketpp::lib::bind;
+
+/* on_open insert connection_hdl into channel
+ * on_close remove connection_hdl from channel
+ * on_message queue send to all channels
+ */
+
+enum action_type {
+    SUBSCRIBE,
+    UNSUBSCRIBE,
+    MESSAGE
+};
+
+struct action {
+    action(action_type t, websocketpp::connection_hdl h) : type(t), hdl(h) {}
+    action(action_type t, websocketpp::connection_hdl h, Server::message_ptr m)
+      : type(t), hdl(h), msg(m) {}
+
+    action_type type;
+    websocketpp::connection_hdl hdl;
+    Server::message_ptr msg;
+};
 
 class Websocket
 {
 public:
-    Websocket(uint16_t port, std::function<std::string(std::string)> on_message);
+    Websocket(uint16_t port, std::function<std::string(std::string)> on_message, std::function<std::string()> on_update);
     void run();
     void stop();
+    void process_messages();
 private:
-    void on_message(Server* s, websocketpp::connection_hdl hdl, Server::message_ptr msg);   
-    Server server; 
-    std::function<std::string(std::string)> p_on_message;
-    uint16_t p_port;
+    void on_open(websocketpp::connection_hdl hdl);
+    void on_close(websocketpp::connection_hdl hdl);
+    void on_message(websocketpp::connection_hdl hdl, Server::message_ptr msg);
+private:
+    std::function<std::string(std::string)> m_on_message;
+    std::function<std::string()> m_on_update;
+    Server m_server; 
+    uint16_t m_port;
+    typedef std::set<websocketpp::connection_hdl,std::owner_less<websocketpp::connection_hdl>> con_list;
+    con_list m_connections;
+
+    std::queue<action> m_actions;
+
+    std::mutex m_action_lock;
+    std::mutex m_connection_lock;
+    std::condition_variable m_action_cond;
 };