websocket

split broadcast out of message function
This commit is contained in:
2024-08-22 11:19:31 +02:00
parent a6ec23fb24
commit aed0d52dbd
3 changed files with 43 additions and 24 deletions
+2
View File
@@ -38,9 +38,11 @@ int main(void)
g_web_socket = &web_socket; g_web_socket = &web_socket;
std::thread message_processor_thread(std::bind(&Websocket::process_messages, &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 // Run the asio loop with the main thread
web_socket.run(); web_socket.run();
message_processor_thread.join(); message_processor_thread.join();
broadcast_processor_thred.join();
} }
+27 -11
View File
@@ -1,7 +1,8 @@
#include "websocket.hpp" #include "websocket.hpp"
Websocket::Websocket(uint16_t port,
Websocket::Websocket(uint16_t port, std::function<std::string(std::string)> on_message, std::function<std::string()> on_update) 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) : m_on_message(on_message), m_on_update(on_update), m_port(port)
{ {
// Set logging settings // Set logging settings
@@ -60,7 +61,8 @@ void Websocket::on_close(websocketpp::connection_hdl hdl)
m_action_cond.notify_one(); 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 // queue message up for sending by processing thread
{ {
std::lock_guard<std::mutex> guard(m_action_lock); std::lock_guard<std::mutex> guard(m_action_lock);
@@ -70,11 +72,14 @@ void Websocket::on_message(websocketpp::connection_hdl hdl, Server::message_ptr
m_action_cond.notify_one(); m_action_cond.notify_one();
} }
void Websocket::process_messages() { void Websocket::process_messages()
while(1) { {
while (1)
{
std::unique_lock<std::mutex> lock(m_action_lock); std::unique_lock<std::mutex> lock(m_action_lock);
while(m_actions.empty()) { while (m_actions.empty())
{
m_action_cond.wait(lock); m_action_cond.wait(lock);
} }
@@ -83,8 +88,7 @@ void Websocket::process_messages() {
lock.unlock(); lock.unlock();
switch (a.type) switch (a.type) {
{
case SUBSCRIBE: case SUBSCRIBE:
{ {
std::lock_guard<std::mutex> guard(m_connection_lock); std::lock_guard<std::mutex> guard(m_connection_lock);
@@ -108,13 +112,25 @@ void Websocket::process_messages() {
// undefined. // undefined.
break; break;
} }
}
}
std::string update = m_on_update(); void Websocket::process_broadcast()
if (!update.empty()) {
while (1)
{
{ {
std::lock_guard<std::mutex> guard(m_connection_lock); 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); m_server.send(*it, update, websocketpp::frame::opcode::TEXT);
} }
} }
+1
View File
@@ -39,6 +39,7 @@ public:
void run(); void run();
void stop(); void stop();
void process_messages(); void process_messages();
void process_broadcast();
private: private:
void on_open(websocketpp::connection_hdl hdl); void on_open(websocketpp::connection_hdl hdl);
void on_close(websocketpp::connection_hdl hdl); void on_close(websocketpp::connection_hdl hdl);