ARCHIVE/clio
1
0
Fork 0
mirror of https://github.com/XRPLF/clio.git synced 2025-11-22 12:45:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

listener rebase

Browse Source
This commit is contained in:
Nathan Nichols
2021-06-14 09:48:40 -05:00
committed by CJ Cobb
parent b4743aa7d1
commit 66c90a8768
7 changed files with 567 additions and 339 deletions
Download Patch File Download Diff File Expand all files Collapse all files

330
reporting/server/HttpBase.h Normal file
View File

@@ -0,0 +1,330 @@
//------------------------------------------------------------------------------
/*
This file is part of rippled: https://github.com/ripple/rippled
Copyright (c) 2021 Ripple Labs Inc.
Permission to use, copy, modify, and/or distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL , DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
#ifndef RIPPLE_REPORTING_HTTP_BASE_SESSION_H
#define RIPPLE_REPORTING_HTTP_BASE_SESSION_H
#include <boost/beast/core.hpp>
#include <boost/beast/http.hpp>
#include <boost/beast/ssl.hpp>
#include <boost/beast/version.hpp>
#include <boost/asio/dispatch.hpp>
#include <boost/asio/strand.hpp>
#include <boost/config.hpp>
#include <algorithm>
#include <cstdlib>
#include <functional>
#include <iostream>
#include <memory>
#include <string>
#include <thread>
#include <vector>
namespace http = boost::beast::http;
namespace net = boost::asio;
namespace ssl = boost::asio::ssl;
using tcp = boost::asio::ip::tcp;
static std::string defaultResponse =
"<!DOCTYPE html><html><head><title>"
" Test page for reporting mode</title></head><body><h1>"
" Test</h1><p>This page shows xrpl reporting http(s) "
"connectivity is working.</p></body></html>";
inline void
httpFail(boost::beast::error_code ec, char const* what)
{
// ssl::error::stream_truncated, also known as an SSL "short read",
// indicates the peer closed the connection without performing the
// required closing handshake (for example, Google does this to
// improve performance). Generally this can be a security issue,
// but if your communication protocol is self-terminated (as
// it is with both HTTP and WebSocket) then you may simply
// ignore the lack of close_notify.
//
// https://github.com/boostorg/beast/issues/38
//
// https://security.stackexchange.com/questions/91435/how-to-handle-a-malicious-ssl-tls-shutdown
//
// When a short read would cut off the end of an HTTP message,
// Beast returns the error boost::beast::http::error::partial_message.
// Therefore, if we see a short read here, it has occurred
// after the message has been completed, so it is safe to ignore it.
if(ec == net::ssl::error::stream_truncated)
return;
std::cerr << what << ": " << ec.message() << "\n";
}
bool
validRequest(boost::json::object const& req)
{
if (!req.contains("method") || !req.at("method").is_string())
return false;
if (!req.contains("params"))
return true;
if (!req.at("params").is_array())
return false;
auto array = req.at("params").as_array();
if (array.size() != 1)
return false;
if (!array.at(0).is_object())
return false;
return true;
}
// This function produces an HTTP response for the given
// request. The type of the response object depends on the
// contents of the request, so the interface requires the
// caller to pass a generic lambda for receiving the response.
template<
class Body, class Allocator,
class Send>
void
handle_request(
boost::beast::http::request<Body, boost::beast::http::basic_fields<Allocator>>&& req,
Send&& send,
ReportingETL& etl)
{
auto const response =
[&req](
http::status status,
std::string content_type,
std::string message)
{
http::response<http::string_body> res{status, req.version()};
res.set(http::field::server, "xrpl-reporting-server-v0.0.0");
res.set(http::field::content_type, content_type);
res.keep_alive(req.keep_alive());
res.body() = std::string(message);
res.prepare_payload();
return res;
};
if(req.method() == http::verb::get
&& req.body() == "")
{
send(response(http::status::ok, "text/html", defaultResponse));
return;
}
if(req.method() != http::verb::post)
{
send(response(
http::status::bad_request,
"text/html",
"Expected a POST request"));
return;
}
try
{
BOOST_LOG_TRIVIAL(info) << "Received request: " << req.body();
boost::json::object request;
try
{
request = boost::json::parse(req.body()).as_object();
}
catch (std::runtime_error const& e)
{
send(response(
http::status::bad_request,
"text/html",
"Cannot parse json in body"));
return;
}
if(!validRequest(request))
{
send(response(
http::status::bad_request,
"text/html",
"Malformed request"));
return;
}
boost::json::object wsStyleRequest = request.contains("params")
? request.at("params").as_array().at(0).as_object()
: boost::json::object{};
wsStyleRequest["command"] = request["method"];
std::cout << "Transfromed to ws style stuff" << std::endl;
auto builtResponse = buildResponse(wsStyleRequest, etl, nullptr);
send(response(
http::status::ok,
"application/json",
boost::json::serialize(builtResponse)));
return;
}
catch (std::exception const& e)
{
std::cout << e.what() << std::endl;
send(response(
http::status::internal_server_error,
"text/html",
"Internal server error occurred"
));
return;
}
}
// From Boost Beast examples http_server_flex.cpp
template<class Derived>
class HttpBase
{
// Access the derived class, this is part of
// the Curiously Recurring Template Pattern idiom.
Derived&
derived()
{
return static_cast<Derived&>(*this);
}
struct send_lambda
{
HttpBase& self_;
explicit
send_lambda(HttpBase& self)
: self_(self)
{
}
template<bool isRequest, class Body, class Fields>
void
operator()(http::message<isRequest, Body, Fields>&& msg) const
{
// The lifetime of the message has to extend
// for the duration of the async operation so
// we use a shared_ptr to manage it.
auto sp = std::make_shared<
http::message<isRequest, Body, Fields>>(std::move(msg));
// Store a type-erased version of the shared
// pointer in the class to keep it alive.
self_.res_ = sp;
// Write the response
http::async_write(
self_.derived().stream(),
*sp,
boost::beast::bind_front_handler(
&HttpBase::on_write,
self_.derived().shared_from_this(),
sp->need_eof()));
}
};
http::request<http::string_body> req_;
std::shared_ptr<void> res_;
ReportingETL& etl_;
send_lambda lambda_;
protected:
boost::beast::flat_buffer buffer_;
public:
HttpBase(ReportingETL& etl, boost::beast::flat_buffer buffer)
: etl_(etl)
, lambda_(*this)
, buffer_(std::move(buffer))
{}
void
do_read()
{
// Make the request empty before reading,
// otherwise the operation behavior is undefined.
req_ = {};
// Set the timeout.
boost::beast::get_lowest_layer(derived().stream()).expires_after(std::chrono::seconds(30));
// Read a request
http::async_read(
derived().stream(),
buffer_,
req_,
boost::beast::bind_front_handler(
&HttpBase::on_read,
derived().shared_from_this()));
}
void
on_read(
boost::beast::error_code ec,
std::size_t bytes_transferred)
{
boost::ignore_unused(bytes_transferred);
// This means they closed the connection
if(ec == http::error::end_of_stream)
return derived().do_close();
if(ec)
return httpFail(ec, "read");
// Send the response
handle_request(std::move(req_), lambda_, etl_);
}
void
on_write(
bool close,
boost::beast::error_code ec,
std::size_t bytes_transferred)
{
boost::ignore_unused(bytes_transferred);
if(ec)
return httpFail(ec, "write");
if(close)
{
// This means we should close the connection, usually because
// the response indicated the "Connection: close" semantic.
return derived().do_close();
}
// We're done with the response so delete it
res_ = nullptr;
// Read another request
do_read();
}
};
#endif //RIPPLE_REPORTING_HTTP_BASE_SESSION_H

288
reporting/server/HttpSession.h
View File

@@ -20,192 +20,34 @@
#ifndef RIPPLE_REPORTING_HTTP_SESSION_H #ifndef RIPPLE_REPORTING_HTTP_SESSION_H
#define RIPPLE_REPORTING_HTTP_SESSION_H #define RIPPLE_REPORTING_HTTP_SESSION_H
#include <boost/beast/core.hpp> #include <reporting/server/HttpBase.h>
#include <boost/beast/http.hpp>
#include <boost/beast/ssl.hpp>
#include <boost/beast/version.hpp>
#include <boost/asio/dispatch.hpp>
#include <boost/asio/strand.hpp>
#include <boost/config.hpp>
#include <algorithm>
#include <cstdlib>
#include <functional>
#include <iostream>
#include <memory>
#include <string>
#include <thread>
#include <vector>
namespace http = boost::beast::http; namespace http = boost::beast::http;
namespace net = boost::asio; namespace net = boost::asio;
namespace ssl = boost::asio::ssl; namespace ssl = boost::asio::ssl;
using tcp = boost::asio::ip::tcp; using tcp = boost::asio::ip::tcp;
static std::string defaultResponse =
"<!DOCTYPE html><html><head><title>"
" Test page for reporting mode</title></head><body><h1>"
" Test</h1><p>This page shows xrpl reporting http(s) "
"connectivity is working.</p></body></html>";
inline void
httpFail(boost::beast::error_code ec, char const* what)
{
// ssl::error::stream_truncated, also known as an SSL "short read",
// indicates the peer closed the connection without performing the
// required closing handshake (for example, Google does this to
// improve performance). Generally this can be a security issue,
// but if your communication protocol is self-terminated (as
// it is with both HTTP and WebSocket) then you may simply
// ignore the lack of close_notify.
//
// https://github.com/boostorg/beast/issues/38
//
// https://security.stackexchange.com/questions/91435/how-to-handle-a-malicious-ssl-tls-shutdown
//
// When a short read would cut off the end of an HTTP message,
// Beast returns the error boost::beast::http::error::partial_message.
// Therefore, if we see a short read here, it has occurred
// after the message has been completed, so it is safe to ignore it.
if(ec == net::ssl::error::stream_truncated)
return;
std::cerr << what << ": " << ec.message() << "\n";
}
// This function produces an HTTP response for the given
// request. The type of the response object depends on the
// contents of the request, so the interface requires the
// caller to pass a generic lambda for receiving the response.
template<
class Body, class Allocator,
class Send>
void
handle_request(
boost::beast::http::request<Body, boost::beast::http::basic_fields<Allocator>>&& req,
Send&& send,
ReportingETL& etl)
{
auto const response =
[&req](
http::status status,
std::string content_type,
std::string message)
{
http::response<http::string_body> res{status, req.version()};
res.set(http::field::server, "xrpl-reporting-server-v0.0.0");
res.set(http::field::content_type, content_type);
res.keep_alive(req.keep_alive());
res.body() = std::string(message);
res.prepare_payload();
return res;
};
if(req.method() == http::verb::get
&& req.body() == "")
{
send(response(http::status::ok, "text/html", defaultResponse));
return;
}
if(req.method() != http::verb::post)
{
send(response(
http::status::bad_request,
"text/html",
"Expected a POST request"));
return;
}
try
{
std::cout << "GOT BODY: " << req.body() << std::endl;
auto request = boost::json::parse(req.body()).as_object();
std::cout << "GOT REQUEST: " << request << std::endl;
auto builtResponse = buildResponse(request, etl, nullptr);
send(response(
http::status::ok,
"application/json",
boost::json::serialize(builtResponse)));
return;
}
catch (std::exception const& e)
{
std::cout << e.what() << std::endl;
send(response(
http::status::internal_server_error,
"text/html",
"Internal server error occurred"
));
return;
}
}
// Handles an HTTP server connection // Handles an HTTP server connection
class HttpSession : public std::enable_shared_from_this<HttpSession> class HttpSession : public HttpBase<HttpSession>
, public std::enable_shared_from_this<HttpSession>
{ {
struct send_lambda boost::beast::tcp_stream stream_;
{
HttpSession& self_;
explicit
send_lambda(HttpSession& self)
: self_(self)
{
}
template<bool isRequest, class Body, class Fields>
void
operator()(http::message<isRequest, Body, Fields>&& msg) const
{
// The lifetime of the message has to extend
// for the duration of the async operation so
// we use a shared_ptr to manage it.
auto sp = std::make_shared<
http::message<isRequest, Body, Fields>>(std::move(msg));
// Store a type-erased version of the shared
// pointer in the class to keep it alive.
self_.res_ = sp;
// Write the response
http::async_write(
self_.stream_,
*sp,
boost::beast::bind_front_handler(
&HttpSession::on_write,
self_.shared_from_this(),
sp->need_eof()));
}
};
boost::beast::ssl_stream<boost::beast::tcp_stream> stream_;
boost::beast::flat_buffer buffer_;
http::request<http::string_body> req_;
std::shared_ptr<void> res_;
send_lambda lambda_;
ReportingETL& etl_;
public: public:
// Take ownership of the socket // Take ownership of the socket
explicit explicit
HttpSession( HttpSession(
tcp::socket&& socket, tcp::socket&& socket,
ssl::context& ctx, ReportingETL& etl,
ReportingETL& etl) boost::beast::flat_buffer buffer)
: stream_(std::move(socket), ctx) : HttpBase<HttpSession>(etl, std::move(buffer))
, lambda_(*this) , stream_(std::move(socket))
, etl_(etl) {}
boost::beast::tcp_stream&
stream()
{ {
return stream_;
} }
// Start the asynchronous operation // Start the asynchronous operation
@@ -219,112 +61,16 @@ public:
net::dispatch( net::dispatch(
stream_.get_executor(), stream_.get_executor(),
boost::beast::bind_front_handler( boost::beast::bind_front_handler(
&HttpSession::on_run, &HttpBase::do_read,
shared_from_this())); shared_from_this()));
} }
void
on_run()
{
// Set the timeout.
boost::beast::get_lowest_layer(stream_).expires_after(
std::chrono::seconds(30));
// Perform the SSL handshake
stream_.async_handshake(
ssl::stream_base::server,
boost::beast::bind_front_handler(
&HttpSession::on_handshake,
shared_from_this()));
}
void
on_handshake(boost::beast::error_code ec)
{
if(ec)
return httpFail(ec, "handshake");
do_read();
}
void
do_read()
{
// Make the request empty before reading,
// otherwise the operation behavior is undefined.
req_ = {};
// Set the timeout.
boost::beast::get_lowest_layer(stream_).expires_after(std::chrono::seconds(30));
// Read a request
http::async_read(stream_, buffer_, req_,
boost::beast::bind_front_handler(
&HttpSession::on_read,
shared_from_this()));
}
void
on_read(
boost::beast::error_code ec,
std::size_t bytes_transferred)
{
boost::ignore_unused(bytes_transferred);
// This means they closed the connection
if(ec == http::error::end_of_stream)
return do_close();
if(ec)
return httpFail(ec, "read");
// Send the response
handle_request(std::move(req_), lambda_, etl_);
}
void
on_write(
bool close,
boost::beast::error_code ec,
std::size_t bytes_transferred)
{
boost::ignore_unused(bytes_transferred);
if(ec)
return httpFail(ec, "write");
if(close)
{
// This means we should close the connection, usually because
// the response indicated the "Connection: close" semantic.
return do_close();
}
// We're done with the response so delete it
res_ = nullptr;
// Read another request
do_read();
}
void void
do_close() do_close()
{ {
// Set the timeout. // Send a TCP shutdown
boost::beast::get_lowest_layer(stream_).expires_after(std::chrono::seconds(30)); boost::beast::error_code ec;
stream_.socket().shutdown(tcp::socket::shutdown_send, ec);
// Perform the SSL shutdown
stream_.async_shutdown(
boost::beast::bind_front_handler(
&HttpSession::on_shutdown,
shared_from_this()));
}
void
on_shutdown(boost::beast::error_code ec)
{
if(ec)
return httpFail(ec, "shutdown");
// At this point the connection is closed gracefully // At this point the connection is closed gracefully
} }

113
reporting/server/SslHttpSession.h Normal file
View File

@@ -0,0 +1,113 @@
//------------------------------------------------------------------------------
/*
This file is part of rippled: https://github.com/ripple/rippled
Copyright (c) 2021 Ripple Labs Inc.
Permission to use, copy, modify, and/or distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL , DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
#ifndef RIPPLE_REPORTING_HTTPS_SESSION_H
#define RIPPLE_REPORTING_HTTPS_SESSION_H
#include <reporting/server/HttpBase.h>
namespace http = boost::beast::http;
namespace net = boost::asio;
namespace ssl = boost::asio::ssl;
using tcp = boost::asio::ip::tcp;
// Handles an HTTPS server connection
class SslHttpSession : public HttpBase<SslHttpSession>
, public std::enable_shared_from_this<SslHttpSession>
{
boost::beast::ssl_stream<boost::beast::tcp_stream> stream_;
public:
// Take ownership of the socket
explicit
SslHttpSession(
tcp::socket&& socket,
ssl::context& ctx,
ReportingETL& etl,
boost::beast::flat_buffer buffer)
: HttpBase<SslHttpSession>(etl, std::move(buffer))
, stream_(std::move(socket), ctx)
{}
boost::beast::ssl_stream<boost::beast::tcp_stream>&
stream()
{
return stream_;
}
// Start the asynchronous operation
void
run()
{
auto self = shared_from_this();
// We need to be executing within a strand to perform async operations
// on the I/O objects in this session.
net::dispatch(stream_.get_executor(), [self]() {
// Set the timeout.
boost::beast::get_lowest_layer(self->stream()).expires_after(
std::chrono::seconds(30));
// Perform the SSL handshake
// Note, this is the buffered version of the handshake.
self->stream_.async_handshake(
ssl::stream_base::server,
self->buffer_.data(),
boost::beast::bind_front_handler(
&SslHttpSession::on_handshake,
self));
});
}
void
on_handshake(
boost::beast::error_code ec,
std::size_t bytes_used)
{
if(ec)
return httpFail(ec, "handshake");
buffer_.consume(bytes_used);
do_read();
}
void
do_close()
{
// Set the timeout.
boost::beast::get_lowest_layer(stream_).expires_after(std::chrono::seconds(30));
// Perform the SSL shutdown
stream_.async_shutdown(
boost::beast::bind_front_handler(
&SslHttpSession::on_shutdown,
shared_from_this()));
}
void
on_shutdown(boost::beast::error_code ec)
{
if(ec)
return httpFail(ec, "shutdown");
// At this point the connection is closed gracefully
}
};
#endif // RIPPLE_REPORTING_HTTPS_SESSION_H

0
reporting/server/WsBase.h Normal file
View File

0
reporting/server/WssSession.h Normal file
View File

161
server/listener.h
View File

@@ -30,84 +30,126 @@
class SubscriptionManager; class SubscriptionManager;
// Accepts incoming connections and launches the sessions // Detects SSL handshakes
template <class Session> class detect_session : public std::enable_shared_from_this<detect_session>
class Listener : public std::enable_shared_from_this<Listener<Session>>
{ {
using std::enable_shared_from_this<Listener<Session>>::shared_from_this; boost::beast::tcp_stream stream_;
std::optional<ssl::context>& ctx_;
boost::asio::io_context& ioc_; ReportingETL& etl_;
boost::asio::ip::tcp::acceptor acceptor_; boost::beast::flat_buffer buffer_;
std::shared_ptr<BackendInterface> backend_;
std::shared_ptr<SubscriptionManager> subscriptions_;
std::shared_ptr<ETLLoadBalancer> balancer_;
DOSGuard& dosGuard_;
ssl::context& ctx_;
public: public:
static void detect_session(
make_listener( tcp::socket&& socket,
boost::asio::io_context& ioc, std::optional<ssl::context>& ctx,
boost::asio::ip::tcp::endpoint endpoint, ReportingETL& etl)
std::shared_ptr<BackendInterface> backend, : stream_(std::move(socket))
std::shared_ptr<SubscriptionManager> subscriptions, , ctx_(ctx)
std::shared_ptr<ETLLoadBalancer> balancer, , etl_(etl)
DOSGuard& dosGuard)
{ {
std::make_shared<listener>(
ioc, endpoint, backend, subscriptions, balancer, dosGuard)
->run();
} }
listener( // Launch the detector
boost::asio::io_context& ioc, void
boost::asio::ip::tcp::endpoint endpoint, run()
std::shared_ptr<BackendInterface> backend, {
std::shared_ptr<SubscriptionManager> subscriptions, // Set the timeout.
std::shared_ptr<ETLLoadBalancer> balancer, boost::beast::get_lowest_layer(stream_).expires_after(std::chrono::seconds(30));
DOSGuard& dosGuard)
if (!ctx_)
{
// Launch plain session
std::make_shared<HttpSession>(
stream_.release_socket(),
etl_,
std::move(buffer_))->run();
}
// Detect a TLS handshake
async_detect_ssl(
stream_,
buffer_,
boost::beast::bind_front_handler(
&detect_session::on_detect,
shared_from_this()));
}
void
on_detect(boost::beast::error_code ec, bool result)
{
if(ec)
return httpFail(ec, "detect");
if(result)
{
// Launch SSL session
std::make_shared<SslHttpSession>(
stream_.release_socket(),
*ctx_,
etl_,
std::move(buffer_))->run();
return;
}
// Launch plain session
std::make_shared<HttpSession>(
stream_.release_socket(),
etl_,
std::move(buffer_))->run();
}
};
// Accepts incoming connections and launches the sessions
class Listener : public std::enable_shared_from_this<Listener>
{
net::io_context& ioc_;
std::optional<ssl::context>& ctx_;
tcp::acceptor acceptor_;
ReportingETL& etl_;
public:
Listener(
net::io_context& ioc,
std::optional<ssl::context>& ctx,
tcp::endpoint endpoint,
ReportingETL& etl)
: ioc_(ioc) : ioc_(ioc)
, acceptor_(ioc) , ctx_(ctx)
, backend_(backend) , acceptor_(net::make_strand(ioc))
, subscriptions_(subscriptions) , etl_(etl)
, balancer_(balancer)
, dosGuard_(dosGuard)
{ {
boost::beast::error_code ec; boost::beast::error_code ec;
// Open the acceptor // Open the acceptor
acceptor_.open(endpoint.protocol(), ec); acceptor_.open(endpoint.protocol(), ec);
if (ec) if(ec)
{ {
BOOST_LOG_TRIVIAL(error) << "Could not open acceptor: " httpFail(ec, "open");
<< ec.message();
return; return;
} }
// Allow address reuse // Allow address reuse
acceptor_.set_option(boost::asio::socket_base::reuse_address(true), ec); acceptor_.set_option(net::socket_base::reuse_address(true), ec);
if (ec) if(ec)
{ {
BOOST_LOG_TRIVIAL(error) << "Could not set option for acceptor: " httpFail(ec, "set_option");
<< ec.message();
return; return;
} }
// Bind to the server address // Bind to the server address
acceptor_.bind(endpoint, ec); acceptor_.bind(endpoint, ec);
if (ec) if(ec)
{ {
BOOST_LOG_TRIVIAL(error) << "Could not bind acceptor: " httpFail(ec, "bind");
<< ec.message();
return; return;
} }
// Start listening for connections // Start listening for connections
acceptor_.listen(boost::asio::socket_base::max_listen_connections, ec); acceptor_.listen(
if (ec) net::socket_base::max_listen_connections, ec);
if(ec)
{ {
BOOST_LOG_TRIVIAL(error) << "Acceptor could not start listening: " httpFail(ec, "listen");
<< ec.message();
return; return;
} }
} }
@@ -126,27 +168,26 @@ private:
{ {
// The new connection gets its own strand // The new connection gets its own strand
acceptor_.async_accept( acceptor_.async_accept(
boost::asio::make_strand(ioc_), net::make_strand(ioc_),
boost::beast::bind_front_handler( boost::beast::bind_front_handler(
&Listener::on_accept, shared_from_this())); &Listener::on_accept,
shared_from_this()));
} }
void void
on_accept(boost::beast::error_code ec, boost::asio::ip::tcp::socket socket) on_accept(boost::beast::error_code ec, tcp::socket socket)
{ {
if (ec) if(ec)
{ {
BOOST_LOG_TRIVIAL(error) << "Failed to accept: " httpFail(ec, "accept");
<< ec.message();
} }
else else
{ {
session::make_session( // Create the detector session and run it
std::make_shared<detect_session>(
std::move(socket), std::move(socket),
backend_, ctx_,
subscriptions_, etl_)->run();
balancer_,
dosGuard_);
} }
// Accept another connection // Accept another connection

14
server/websocket_server_async.cpp
View File

@@ -152,7 +152,7 @@ int
main(int argc, char* argv[]) main(int argc, char* argv[])
{ {
// Check command line arguments. // Check command line arguments.
if (argc != 5 and argc != 6) if (argc < 3 || argc > 6)
{ {
std::cerr std::cerr
<< "Usage: websocket-server-async <threads> " << "Usage: websocket-server-async <threads> "
@@ -164,7 +164,10 @@ main(int argc, char* argv[])
auto const threads = std::max<int>(1, std::atoi(argv[1])); auto const threads = std::max<int>(1, std::atoi(argv[1]));
auto const config = parse_config(argv[2]); auto const config = parse_config(argv[2]);
auto ctx = parse_certs(argv[3], argv[4]);
std::optional<ssl::context> ctx = {};
if (argc == 4 || argc == 5)
ctx = parse_certs(argv[3], argv[4]);
if (argc > 5) if (argc > 5)
{ {
@@ -174,17 +177,12 @@ main(int argc, char* argv[])
{ {
initLogLevel(2); initLogLevel(2);
} }
if (!config) if (!config)
{ {
std::cerr << "couldnt parse config. Exiting..." << std::endl; std::cerr << "couldnt parse config. Exiting..." << std::endl;
return EXIT_FAILURE; return EXIT_FAILURE;
} }
if (!ctx)
{
std::cerr << "could not parse certs, Exiting..." << std::endl;
return EXIT_FAILURE;
}
boost::asio::io_context ioc{threads}; boost::asio::io_context ioc{threads};