diff --git a/tools/main.cpp b/tools/main.cpp index 6c3436e..b907420 100644 --- a/tools/main.cpp +++ b/tools/main.cpp @@ -1,7 +1,6 @@ #include #include #include -#include #include #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(); } \ No newline at end of file diff --git a/tools/websocket.cpp b/tools/websocket.cpp index a5c2434..2e54e7a 100644 --- a/tools/websocket.cpp +++ b/tools/websocket.cpp @@ -1,38 +1,32 @@ #include "websocket.hpp" -Websocket::Websocket(uint16_t port, std::function on_message) - : p_on_message(on_message), p_port(port) +Websocket::Websocket(uint16_t port, std::function on_message, std::function 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 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 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 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 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 guard(m_connection_lock); + m_connections.insert(a.hdl); + } + break; + case UNSUBSCRIBE: + { + std::lock_guard guard(m_connection_lock); + m_connections.erase(a.hdl); + } + break; + case MESSAGE: + { + std::lock_guard 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 guard(m_connection_lock); + for (auto it = m_connections.begin(); it != m_connections.end(); ++it) + { + m_server.send(*it, update, websocketpp::frame::opcode::TEXT); + } + } } } \ No newline at end of file diff --git a/tools/websocket.hpp b/tools/websocket.hpp index e32c6d8..0109f79 100644 --- a/tools/websocket.hpp +++ b/tools/websocket.hpp @@ -1,6 +1,7 @@ #pragma once #include +#include #include #include @@ -9,17 +10,50 @@ typedef websocketpp::server 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 on_message); + Websocket(uint16_t port, std::function on_message, std::function 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 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 m_on_message; + std::function m_on_update; + Server m_server; + uint16_t m_port; + typedef std::set> con_list; + con_list m_connections; + + std::queue m_actions; + + std::mutex m_action_lock; + std::mutex m_connection_lock; + std::condition_variable m_action_cond; }; \ No newline at end of file