summaryrefslogtreecommitdiff
path: root/graphics/asymptote/LspCpp/src/jsonrpc/TcpServer.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'graphics/asymptote/LspCpp/src/jsonrpc/TcpServer.cpp')
-rwxr-xr-xgraphics/asymptote/LspCpp/src/jsonrpc/TcpServer.cpp296
1 files changed, 296 insertions, 0 deletions
diff --git a/graphics/asymptote/LspCpp/src/jsonrpc/TcpServer.cpp b/graphics/asymptote/LspCpp/src/jsonrpc/TcpServer.cpp
new file mode 100755
index 0000000000..37a7a3a3af
--- /dev/null
+++ b/graphics/asymptote/LspCpp/src/jsonrpc/TcpServer.cpp
@@ -0,0 +1,296 @@
+//
+// server.cpp
+
+#include "LibLsp/JsonRpc/TcpServer.h"
+#include <signal.h>
+#include <utility>
+#include <boost/bind/bind.hpp>
+
+#include "LibLsp/JsonRpc/MessageIssue.h"
+#include "LibLsp/JsonRpc/stream.h"
+
+
+namespace lsp {
+ struct tcp_connect_session;
+
+
+ class tcp_stream_wrapper :public istream, public ostream
+ {
+ public:
+ tcp_stream_wrapper(tcp_connect_session& _w);
+
+ tcp_connect_session& session;
+ std::atomic<bool> quit{};
+ std::shared_ptr < MultiQueueWaiter> request_waiter;
+ ThreadedQueue< char > on_request;
+ std::string error_message;
+
+
+ bool fail() override
+ {
+ return bad();
+ }
+
+
+
+ bool eof() override
+ {
+ return bad();
+ }
+ bool good() override
+ {
+ return !bad();
+ }
+ tcp_stream_wrapper& read(char* str, std::streamsize count)
+ override
+ {
+ auto some = on_request.TryDequeueSome(static_cast<size_t>( count ));
+ memcpy(str,some.data(),some.size());
+ for (std::streamsize i = some.size(); i < count; ++i)
+ {
+ str[i] = static_cast<char>(get());
+ }
+
+ return *this;
+ }
+ int get() override
+ {
+ return on_request.Dequeue();
+ }
+
+ bool bad() override;
+
+ tcp_stream_wrapper& write(const std::string& c) override;
+
+ tcp_stream_wrapper& write(std::streamsize _s) override;
+
+ tcp_stream_wrapper& flush() override
+ {
+ return *this;
+ }
+ void reset_state()
+ {
+ return;
+ }
+
+ void clear() override
+ {
+
+ }
+
+ std::string what() override;
+ bool need_to_clear_the_state() override
+ {
+ return false;
+ }
+ };
+ struct tcp_connect_session:std::enable_shared_from_this<tcp_connect_session>
+ {
+ /// Buffer for incoming data.
+ std::array<unsigned char, 8192> buffer_;
+ boost::asio::ip::tcp::socket socket_;
+ /// Strand to ensure the connection's handlers are not called concurrently.
+ boost::asio::io_context::strand strand_;
+ std::shared_ptr<tcp_stream_wrapper> proxy_;
+ explicit tcp_connect_session(boost::asio::io_context& io_context, boost::asio::ip::tcp::socket&& _socket)
+ : socket_(std::move(_socket)), strand_(io_context), proxy_(new tcp_stream_wrapper(*this))
+ {
+ do_read();
+ }
+ void do_write(const std::string& data)
+ {
+ socket_.async_write_some(boost::asio::buffer(data.data(), data.size()),
+ boost::asio::bind_executor(strand_,[this](boost::system::error_code ec, std::size_t n)
+ {
+ if (!ec)
+ {
+ return;
+ }
+ proxy_->error_message = ec.message();
+
+ }));
+ }
+ void do_read()
+ {
+ socket_.async_read_some(boost::asio::buffer(buffer_),
+ boost::asio::bind_executor(strand_,
+ [this](boost::system::error_code ec, size_t bytes_transferred)
+ {
+ if (!ec)
+ {
+ std::vector<char> elements(buffer_.data(), buffer_.data() + bytes_transferred);
+ proxy_->on_request.EnqueueAll(std::move(elements), false);
+ do_read();
+ return;
+ }
+ proxy_->error_message = ec.message();
+
+ }));
+ }
+ };
+
+ tcp_stream_wrapper::tcp_stream_wrapper(tcp_connect_session& _w): session(_w)
+ {
+ }
+
+ bool tcp_stream_wrapper::bad()
+ {
+ return !session.socket_.is_open();
+ }
+
+ tcp_stream_wrapper& tcp_stream_wrapper::write(const std::string& c)
+ {
+ session.do_write(c);
+ return *this;
+ }
+
+ tcp_stream_wrapper& tcp_stream_wrapper::write(std::streamsize _s)
+ {
+ session.do_write(std::to_string(_s));
+ return *this;
+ }
+
+ std::string tcp_stream_wrapper::what()
+ {
+ if (error_message.size())
+ return error_message;
+
+ if(! session.socket_.is_open())
+ {
+ return "Socket is not open.";
+ }
+ return {};
+ }
+
+ struct TcpServer::Data
+ {
+ Data(
+ lsp::Log& log, uint32_t _max_workers) :
+ acceptor_(io_context_), _log(log)
+ {
+ }
+
+ ~Data()
+ {
+
+ }
+ /// The io_context used to perform asynchronous operations.
+ boost::asio::io_context io_context_;
+
+ std::shared_ptr<boost::asio::io_service::work> work;
+
+ std::shared_ptr<tcp_connect_session> _connect_session;
+ /// Acceptor used to listen for incoming connections.
+ boost::asio::ip::tcp::acceptor acceptor_;
+
+ lsp::Log& _log;
+
+ };
+
+ TcpServer::~TcpServer()
+ {
+ delete d_ptr;
+ }
+
+ TcpServer::TcpServer(const std::string& address, const std::string& port,
+ std::shared_ptr < MessageJsonHandler> json_handler,
+ std::shared_ptr < Endpoint> localEndPoint, lsp::Log& log, uint32_t _max_workers)
+ : point(json_handler, localEndPoint, log, _max_workers),d_ptr(new Data( log, _max_workers))
+
+ {
+
+ d_ptr->work = std::make_shared<boost::asio::io_service::work>(d_ptr->io_context_);
+
+ // Open the acceptor with the option to reuse the address (i.e. SO_REUSEADDR).
+ boost::asio::ip::tcp::resolver resolver(d_ptr->io_context_);
+ boost::asio::ip::tcp::endpoint endpoint =
+ *resolver.resolve(address, port).begin();
+ d_ptr->acceptor_.open(endpoint.protocol());
+ d_ptr->acceptor_.set_option(boost::asio::ip::tcp::acceptor::reuse_address(true));
+ try
+ {
+ d_ptr->acceptor_.bind(endpoint);
+ }
+ catch (boost::system::system_error & e)
+ {
+ std::string temp = "Socket Server bind failed.";
+ d_ptr->_log.log(lsp::Log::Level::INFO , temp + e.what());
+ return;
+ }
+ d_ptr->acceptor_.listen();
+
+ do_accept();
+ std::string desc = "Socket TcpServer " + address + " " + port + " start.";
+ d_ptr->_log.log(lsp::Log::Level::INFO, desc);
+ }
+
+ void TcpServer::run()
+ {
+ // The io_context::run() call will block until all asynchronous operations
+ // have finished. While the TcpServer is running, there is always at least one
+ // asynchronous operation outstanding: the asynchronous accept call waiting
+ // for new incoming connections.
+ d_ptr->io_context_.run();
+
+ }
+
+ void TcpServer::stop()
+ {
+ try
+ {
+ if(d_ptr->work)
+ d_ptr->work.reset();
+
+ do_stop();
+ }
+ catch (...)
+ {
+ }
+ }
+
+ void TcpServer::do_accept()
+ {
+ d_ptr->acceptor_.async_accept(
+ [this](boost::system::error_code ec, boost::asio::ip::tcp::socket socket)
+ {
+ // Check whether the TcpServer was stopped by a signal before this
+ // completion handler had a chance to run.
+ if (!d_ptr->acceptor_.is_open())
+ {
+ return;
+ }
+
+ if (!ec)
+ {
+ if(d_ptr->_connect_session)
+ {
+ if(d_ptr->_connect_session->socket_.is_open())
+ {
+ std::string desc = "Disconnect previous client " + d_ptr->_connect_session->socket_.local_endpoint().address().to_string();
+ d_ptr->_log.log(lsp::Log::Level::INFO, desc);
+ d_ptr->_connect_session->socket_.close();
+ }
+
+ point.Stop();
+ }
+ auto local_point = socket.local_endpoint();
+
+ std::string desc = ("New client " + local_point.address().to_string() + " connect.");
+ d_ptr->_log.log(lsp::Log::Level::INFO, desc);
+ d_ptr->_connect_session = std::make_shared<tcp_connect_session>(d_ptr->io_context_,std::move(socket));
+
+ point.startProcessingMessages(d_ptr->_connect_session->proxy_, d_ptr->_connect_session->proxy_);
+ do_accept();
+ }
+ });
+ }
+
+ void TcpServer::do_stop()
+ {
+ d_ptr->acceptor_.close();
+
+ point.Stop();
+
+ }
+
+ } // namespace