diff options
Diffstat (limited to 'src/osd/modules/output/network.cpp')
-rw-r--r-- | src/osd/modules/output/network.cpp | 153 |
1 files changed, 119 insertions, 34 deletions
diff --git a/src/osd/modules/output/network.cpp b/src/osd/modules/output/network.cpp index 712d0d4c6c5..54b00129ff0 100644 --- a/src/osd/modules/output/network.cpp +++ b/src/osd/modules/output/network.cpp @@ -6,63 +6,147 @@ Network output interface. -*******************************************************************c********/ +***************************************************************************/ #include "output_module.h" #include "modules/osdmodule.h" -#include "modules/ipc/raw_tcp_server.h" -#include "modules/ipc/raw_tcp_connection.h" -#include <uv.h> -#include <mutex> #include <thread> +#include <set> +#include "asio.h" -class output_network_server : - public raw_tcp_server::listener, - public raw_tcp_connection::listener +class output_client { public: - output_network_server(uv_loop_t* loop) { m_tcp_server = new raw_tcp_server(loop, "0.0.0.0", 8000, 256, this, this); } + virtual ~output_client() {} + virtual void deliver(std::string &msg) = 0; +}; - virtual ~output_network_server() { delete m_tcp_server; } +using output_client_ptr = std::shared_ptr<output_client>; +using client_set = std::set<output_client_ptr>; - void terminate_all() const { m_tcp_server->terminate(); } - void send_to_all(const uint8_t* data, size_t len) const { m_tcp_server->send_to_all(data,len); } +class output_session : public output_client, + public std::enable_shared_from_this<output_session> + +{ +public: + output_session(asio::ip::tcp::socket socket, client_set *clients) : + m_socket(std::move(socket)), + m_clients(clients) + { + } + + void start() + { + m_clients->insert(shared_from_this()); + do_read(); + } - /* Pure virtual methods inherited from raw_tcp_server::listener. */ - virtual void on_raw_tcp_connection_closed(raw_tcp_server* tcpServer, raw_tcp_connection* connection, bool is_closed_by_peer) override { } - /* Pure virtual methods inherited from raw_tcp_connection::listener. */ - virtual void on_data_recv(raw_tcp_connection *connection, const uint8_t* data, size_t len) override { } private: - raw_tcp_server* m_tcp_server; + void deliver(std::string &msg) + { + std::strncpy(m_data, msg.c_str(), max_length); + do_write(msg.size()); + } + + void do_read() + { + auto self(shared_from_this()); + m_socket.async_read_some(asio::buffer(m_input_m_data, max_length), + [this, self](std::error_code ec, std::size_t length) + { + if (!ec) + { + do_read(); + } + else + { + m_clients->erase(shared_from_this()); + } + }); + } + + void do_write(std::size_t length) + { + auto self(shared_from_this()); + asio::async_write(m_socket, asio::buffer(m_data, length), + [this, self](std::error_code ec, std::size_t /*length*/) + { + if (ec) + { + m_clients->erase(shared_from_this()); + } + }); + } + + asio::ip::tcp::socket m_socket; + enum { max_length = 1024 }; + char m_data[max_length]; + char m_input_m_data[max_length]; + client_set *m_clients; }; +class output_network_server +{ +public: + output_network_server(asio::io_context& io_context, short port) : + m_acceptor(io_context, asio::ip::tcp::endpoint(asio::ip::tcp::v4(), port)) + { + do_accept(); + } + + void deliver_to_all(std::string msg) + { + for (auto client: m_clients) + client->deliver(msg); + } + +private: + void do_accept() + { + m_acceptor.async_accept( + [this](std::error_code ec, asio::ip::tcp::socket socket) + { + if (!ec) + { + std::make_shared<output_session>(std::move(socket),&m_clients)->start(); + } + + do_accept(); + }); + } + + asio::ip::tcp::acceptor m_acceptor; + client_set m_clients; +}; + + class output_network : public osd_module, public output_module { public: output_network() - : osd_module(OSD_OUTPUT_PROVIDER, "network"), output_module(), m_loop(nullptr), m_server(nullptr) -{ + : osd_module(OSD_OUTPUT_PROVIDER, "network"), + output_module(), + m_io_context(nullptr), m_server(nullptr) + { } - virtual ~output_network() { + + virtual ~output_network() + { } - virtual int init(const osd_options &options) override { - m_loop = new uv_loop_t; - int err = uv_loop_init(m_loop); - if (err) { - return 1; - } + virtual int init(const osd_options &options) override + { m_working_thread = std::thread([](output_network* self) { self->process_output(); }, this); return 0; } - virtual void exit() override { - m_server->terminate_all(); + virtual void exit() override + { + m_io_context->stop(); m_working_thread.join(); - uv_loop_close(m_loop); - delete m_loop; delete m_server; + delete m_io_context; } // output_module @@ -71,19 +155,20 @@ public: { static char buf[256]; sprintf(buf, "%s = %d\n", ((outname==nullptr) ? "none" : outname), value); - m_server->send_to_all((const uint8_t*)buf, strlen(buf)); + m_server->deliver_to_all(buf); } // implementation void process_output() { - m_server = new output_network_server(m_loop); - uv_run(m_loop, UV_RUN_DEFAULT); + m_io_context = new asio::io_context(); + m_server = new output_network_server(*m_io_context, 8000); + m_io_context->run(); } private: std::thread m_working_thread; - uv_loop_t* m_loop; + asio::io_context *m_io_context; output_network_server *m_server; }; |