summaryrefslogtreecommitdiff
path: root/graphics/asymptote/LspCpp/src/jsonrpc/WebSocketServer.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'graphics/asymptote/LspCpp/src/jsonrpc/WebSocketServer.cpp')
-rwxr-xr-xgraphics/asymptote/LspCpp/src/jsonrpc/WebSocketServer.cpp337
1 files changed, 337 insertions, 0 deletions
diff --git a/graphics/asymptote/LspCpp/src/jsonrpc/WebSocketServer.cpp b/graphics/asymptote/LspCpp/src/jsonrpc/WebSocketServer.cpp
new file mode 100755
index 0000000000..127cf2a9f4
--- /dev/null
+++ b/graphics/asymptote/LspCpp/src/jsonrpc/WebSocketServer.cpp
@@ -0,0 +1,337 @@
+#include "LibLsp/JsonRpc/MessageIssue.h"
+#include "LibLsp/JsonRpc/WebSocketServer.h"
+#include <iostream>
+#include <signal.h>
+#include <utility>
+#include "LibLsp/JsonRpc/stream.h"
+#include <boost/beast/core.hpp>
+#include <boost/beast/websocket.hpp>
+#include <boost/asio/dispatch.hpp>
+namespace beast = boost::beast; // from <boost/beast.hpp>
+namespace http = beast::http; // from <boost/beast/http.hpp>
+namespace websocket = beast::websocket; // from <boost/beast/websocket.hpp>
+namespace net = boost::asio; // from <boost/asio.hpp>
+using tcp = boost::asio::ip::tcp; // from <boost/asio/ip/tcp.hpp>
+namespace lsp {
+
+ // Echoes back all received WebSocket messages
+ class server_session : public std::enable_shared_from_this<server_session>
+ {
+ websocket::stream<beast::tcp_stream> ws_;
+
+ beast::flat_buffer buffer_;
+ std::string user_agent_;
+ public:
+ std::shared_ptr<websocket_stream_wrapper> proxy_;
+ // Take ownership of the socket
+ explicit
+ server_session(tcp::socket&& socket,const std::string& user_agent)
+ : ws_(std::move(socket)),user_agent_(user_agent)
+ {
+ proxy_ = std::make_shared<websocket_stream_wrapper>(ws_);
+ }
+
+ // Get on the correct executor
+ void
+ run()
+ {
+ // We need to be executing within a strand to perform async operations
+ // on the I/O objects in this server_session. Although not strictly necessary
+ // for single-threaded contexts, this example code is written to be
+ // thread-safe by default.
+ net::dispatch(ws_.get_executor(),
+ beast::bind_front_handler(
+ &server_session::on_run,
+ shared_from_this()));
+ }
+
+ // Start the asynchronous operation
+ void
+ on_run()
+ {
+ // Set suggested timeout settings for the websocket
+ ws_.set_option(
+ websocket::stream_base::timeout::suggested(
+ beast::role_type::server));
+
+ // Set a decorator to change the Server of the handshake
+ ws_.set_option(websocket::stream_base::decorator(
+ [=](websocket::response_type& res)
+ {
+ res.set(http::field::server, user_agent_.c_str());
+ }));
+ // Accept the websocket handshake
+ ws_.async_accept(
+ beast::bind_front_handler(
+ &server_session::on_accept,
+ shared_from_this()));
+ }
+
+ void
+ on_accept(beast::error_code ec)
+ {
+ if (ec)
+ return ;
+
+ // Read a message
+ // Read a message into our buffer
+ ws_.async_read(
+ buffer_,
+ beast::bind_front_handler(
+ &server_session::on_read,
+ shared_from_this()));
+ }
+
+
+
+ void
+ on_read(
+ beast::error_code ec,
+ std::size_t bytes_transferred)
+ {
+
+ if(!ec)
+ {
+ char* data = reinterpret_cast<char*>(buffer_.data().data());
+ std::vector<char> elements(data, data + bytes_transferred);
+
+ buffer_.clear();
+ proxy_->on_request.EnqueueAll(std::move(elements), false);
+
+ // Read a message into our buffer
+ ws_.async_read(
+ buffer_,
+ beast::bind_front_handler(
+ &server_session::on_read,
+ shared_from_this()));
+ return;
+ }
+ if (ec){
+ proxy_->error_message = ec.message();
+ }
+ }
+
+
+
+ void close()
+ {
+ if(ws_.is_open())
+ {
+ boost::system::error_code ec;
+ ws_.close(websocket::close_code::normal, ec);
+ }
+
+ }
+ };
+
+ //------------------------------------------------------------------------------
+
+ struct WebSocketServer::Data
+ {
+ Data(const std::string& user_agent, lsp::Log& log) :
+ acceptor_(io_context_), user_agent_(user_agent), _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;
+
+ /// Acceptor used to listen for incoming connections.
+ boost::asio::ip::tcp::acceptor acceptor_;
+
+ std::shared_ptr < server_session> _server_session;
+
+ std::string user_agent_;
+ lsp::Log& _log;
+
+ };
+
+ websocket_stream_wrapper::websocket_stream_wrapper(boost::beast::websocket::stream<boost::beast::tcp_stream>& _w):
+ ws_(_w), request_waiter(new MultiQueueWaiter()),
+ on_request(request_waiter)
+ {
+ }
+
+ bool websocket_stream_wrapper::fail()
+ {
+ return bad();
+ }
+
+ bool websocket_stream_wrapper::eof()
+ {
+ return bad();
+ }
+
+ bool websocket_stream_wrapper::good()
+ {
+ return !bad();
+ }
+
+ websocket_stream_wrapper& websocket_stream_wrapper::read(char* str, std::streamsize count)
+ {
+ 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 websocket_stream_wrapper::get()
+ {
+ return on_request.Dequeue();
+ }
+
+ bool websocket_stream_wrapper::bad()
+ {
+ return !ws_.next_layer().socket().is_open();
+ }
+
+ websocket_stream_wrapper& websocket_stream_wrapper::write(const std::string& c)
+ {
+ ws_.write(boost::asio::buffer(std::string(c)));
+ return *this;
+ }
+
+
+ websocket_stream_wrapper& websocket_stream_wrapper::write(std::streamsize _s)
+ {
+ std::ostringstream temp;
+ temp << _s;
+ ws_.write(boost::asio::buffer(temp.str()));
+ return *this;
+ }
+
+ websocket_stream_wrapper& websocket_stream_wrapper::flush()
+ {
+ return *this;
+ }
+
+ void websocket_stream_wrapper::clear()
+ {
+ }
+
+ std::string websocket_stream_wrapper::what()
+ {
+ if (!error_message.empty())
+ return error_message;
+
+ if (!ws_.next_layer().socket().is_open())
+ {
+ return "Socket is not open.";
+ }
+ return {};
+ }
+
+ WebSocketServer::~WebSocketServer()
+ {
+ delete d_ptr;
+ }
+
+ WebSocketServer::WebSocketServer(const std::string& user_agent, 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(user_agent,log))
+
+ {
+
+ 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 blid faild.";
+ d_ptr->_log.log(lsp::Log::Level::INFO , temp + e.what());
+ return;
+ }
+ d_ptr->acceptor_.listen();
+
+ do_accept();
+ std::string desc = "Socket WebSocketServer " + address + " " + port + " start.";
+ d_ptr->_log.log(lsp::Log::Level::INFO, desc);
+ }
+
+ void WebSocketServer::run()
+ {
+ // The io_context::run() call will block until all asynchronous operations
+ // have finished. While the WebSocketServer 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 WebSocketServer::stop()
+ {
+ try
+ {
+ if(d_ptr->work)
+ d_ptr->work.reset();
+
+ do_stop();
+ }
+ catch (...)
+ {
+ }
+ }
+
+ void WebSocketServer::do_accept()
+ {
+ d_ptr->acceptor_.async_accept(
+ [this](boost::system::error_code ec, boost::asio::ip::tcp::socket socket)
+ {
+ // Check whether the WebSocketServer 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->_server_session)
+ {
+ try
+ {
+ d_ptr->_server_session->close();
+ point.Stop();
+ }
+ catch (...)
+ {
+ }
+ }
+ d_ptr->_server_session = std::make_shared<server_session>(std::move(socket), d_ptr->user_agent_);
+ d_ptr->_server_session->run();
+
+ point.startProcessingMessages(d_ptr->_server_session->proxy_, d_ptr->_server_session->proxy_);
+ do_accept();
+ }
+
+ });
+ }
+
+ void WebSocketServer::do_stop()
+ {
+ d_ptr->acceptor_.close();
+
+ point.Stop();
+
+ }
+
+ } // namespace