Fixes to support Asio changes (API Change):

This adjusts Beast's interfaces and implementation to match
the changes in Boost.Asio.
This commit is contained in:
Vinnie Falco
2019-02-19 16:23:30 -08:00
parent ac24e58fb3
commit 6baa607295
58 changed files with 762 additions and 723 deletions
@@ -48,13 +48,13 @@ fail(beast::error_code ec, char const* what)
// Echoes back all received WebSocket messages
class session : public std::enable_shared_from_this<session>
{
websocket::stream<beast::ssl_stream<
beast::tcp_stream<net::io_context::strand>>> ws_;
websocket::stream<
beast::ssl_stream<beast::tcp_stream>> ws_;
beast::multi_buffer buffer_;
public:
// Take ownership of the socket
session(tcp::socket socket, ssl::context& ctx)
session(tcp::socket&& socket, ssl::context& ctx)
: ws_(std::move(socket), ctx)
{
}
@@ -172,16 +172,20 @@ public:
// Accepts incoming connections and launches the sessions
class listener : public std::enable_shared_from_this<listener>
{
net::io_context& ioc_;
ssl::context& ctx_;
tcp::acceptor acceptor_;
tcp::socket socket_;
public:
listener(
net::io_context& ioc,
ssl::context& ctx,
tcp::endpoint endpoint)
: ctx_(ctx)
, acceptor_(ioc)
: ioc_(ioc)
, ctx_(ctx)
, acceptor_(beast::make_strand(ioc))
, socket_(beast::make_strand(ioc))
{
beast::error_code ec;
@@ -232,13 +236,14 @@ public:
do_accept()
{
acceptor_.async_accept(
socket_,
beast::bind_front_handler(
&listener::on_accept,
shared_from_this()));
}
void
on_accept(beast::error_code ec, tcp::socket socket)
on_accept(beast::error_code ec)
{
if(ec)
{
@@ -247,9 +252,12 @@ public:
else
{
// Create the session and run it
std::make_shared<session>(std::move(socket), ctx_)->run();
std::make_shared<session>(std::move(socket_), ctx_)->run();
}
// Make sure each session gets its own strand
socket_ = tcp::socket(beast::make_strand(ioc_));
// Accept another connection
do_accept();
}
@@ -43,13 +43,13 @@ fail(beast::error_code ec, char const* what)
// Echoes back all received WebSocket messages
class session : public std::enable_shared_from_this<session>
{
websocket::stream<beast::tcp_stream<net::io_context::strand>> ws_;
websocket::stream<beast::tcp_stream> ws_;
beast::multi_buffer buffer_;
public:
// Take ownership of the socket
explicit
session(tcp::socket socket)
session(tcp::socket&& socket)
: ws_(std::move(socket))
{
}
@@ -146,13 +146,17 @@ public:
// Accepts incoming connections and launches the sessions
class listener : public std::enable_shared_from_this<listener>
{
net::io_context& ioc_;
tcp::acceptor acceptor_;
tcp::socket socket_;
public:
listener(
net::io_context& ioc,
tcp::endpoint endpoint)
: acceptor_(ioc)
: ioc_(ioc)
, acceptor_(ioc)
, socket_(beast::make_strand(ioc_))
{
beast::error_code ec;
@@ -203,13 +207,14 @@ public:
do_accept()
{
acceptor_.async_accept(
socket_,
beast::bind_front_handler(
&listener::on_accept,
shared_from_this()));
}
void
on_accept(beast::error_code ec, tcp::socket socket)
on_accept(beast::error_code ec)
{
if(ec)
{
@@ -218,9 +223,12 @@ public:
else
{
// Create the session and run it
std::make_shared<session>(std::move(socket))->run();
std::make_shared<session>(std::move(socket_))->run();
}
// Make sure each session gets its own strand
socket_ = tcp::socket(beast::make_strand(ioc_));
// Accept another connection
do_accept();
}
@@ -21,7 +21,7 @@
*/
class http_session : public boost::enable_shared_from_this<http_session>
{
beast::tcp_stream<net::io_context::strand> stream_;
beast::tcp_stream stream_;
beast::flat_buffer buffer_;
boost::shared_ptr<shared_state> state_;
http::request<http::string_body> req_;
@@ -16,7 +16,9 @@ listener(
net::io_context& ioc,
tcp::endpoint endpoint,
boost::shared_ptr<shared_state> const& state)
: acceptor_(ioc)
: ioc_(ioc)
, acceptor_(ioc)
, socket_(beast::make_strand(ioc))
, state_(state)
{
beast::error_code ec;
@@ -61,6 +63,7 @@ run()
{
// Start accepting a connection
acceptor_.async_accept(
socket_,
beast::bind_front_handler(
&listener::on_accept,
shared_from_this()));
@@ -80,18 +83,22 @@ fail(beast::error_code ec, char const* what)
// Handle a connection
void
listener::
on_accept(beast::error_code ec, tcp::socket socket)
on_accept(beast::error_code ec)
{
if(ec)
return fail(ec, "accept");
else
// Launch a new session for this connection
boost::make_shared<http_session>(
std::move(socket),
std::move(socket_),
state_)->run();
// Make sure each session gets its own strand
socket_ = tcp::socket(beast::make_strand(ioc_));
// Accept another connection
acceptor_.async_accept(
socket_,
beast::bind_front_handler(
&listener::on_accept,
shared_from_this()));
@@ -22,11 +22,13 @@ class shared_state;
// Accepts incoming connections and launches the sessions
class listener : public boost::enable_shared_from_this<listener>
{
net::io_context& ioc_;
tcp::acceptor acceptor_;
tcp::socket socket_;
boost::shared_ptr<shared_state> state_;
void fail(beast::error_code ec, char const* what);
void on_accept(beast::error_code ec, tcp::socket socket);
void on_accept(beast::error_code ec);
public:
listener(
@@ -83,17 +83,22 @@ void
websocket_session::
send(boost::shared_ptr<std::string const> const& ss)
{
// Get on the strand if we aren't already,
// otherwise we will concurrently access
// objects which are not thread-safe.
if(! ws_.get_executor().running_in_this_thread())
return net::post(
ws_.get_executor(),
beast::bind_front_handler(
&websocket_session::send,
shared_from_this(),
ss));
// Post our work to the strand, this ensures
// that the members of `this` will not be
// accessed concurrently.
net::post(
ws_.get_executor(),
beast::bind_front_handler(
&websocket_session::on_send,
shared_from_this(),
ss));
}
void
websocket_session::
on_send(boost::shared_ptr<std::string const> const& ss)
{
// Always add to queue
queue_.push_back(ss);
@@ -27,7 +27,7 @@ class shared_state;
class websocket_session : public boost::enable_shared_from_this<websocket_session>
{
beast::flat_buffer buffer_;
websocket::stream<beast::tcp_stream<net::io_context::strand>> ws_;
websocket::stream<beast::tcp_stream> ws_;
boost::shared_ptr<shared_state> state_;
std::vector<boost::shared_ptr<std::string const>> queue_;
@@ -50,6 +50,10 @@ public:
// Send a message
void
send(boost::shared_ptr<std::string const> const& ss);
private:
void
on_send(boost::shared_ptr<std::string const> const& ss);
};
template<class Body, class Allocator>
@@ -38,11 +38,6 @@ using tcp = boost::asio::ip::tcp; // from <boost/asio/ip/tcp.hpp>
//------------------------------------------------------------------------------
// The type of websocket stream to use
// Stackful coroutines are already stranded.
using ws_type = websocket::stream<beast::ssl_stream<
beast::tcp_stream<net::io_context::executor_type>>>;
// Report a failure
void
fail(beast::error_code ec, char const* what)
@@ -53,7 +48,8 @@ fail(beast::error_code ec, char const* what)
// Echoes back all received WebSocket messages
void
do_session(
ws_type& ws,
websocket::stream<
beast::ssl_stream<beast::tcp_stream>>& ws,
net::yield_context yield)
{
beast::error_code ec;
@@ -153,10 +149,11 @@ do_listen(
fail(ec, "accept");
else
net::spawn(
acceptor.get_executor().context(),
acceptor.get_executor(),
std::bind(
&do_session,
ws_type(std::move(socket), ctx),
websocket::stream<beast::ssl_stream<
beast::tcp_stream>>(std::move(socket), ctx),
std::placeholders::_1));
}
}
@@ -33,11 +33,6 @@ using tcp = boost::asio::ip::tcp; // from <boost/asio/ip/tcp.hpp>
//------------------------------------------------------------------------------
// The type of websocket stream to use
// Stackful coroutines are already stranded.
using ws_type = websocket::stream<
beast::tcp_stream<net::io_context::executor_type>>;
// Report a failure
void
fail(beast::error_code ec, char const* what)
@@ -47,7 +42,9 @@ fail(beast::error_code ec, char const* what)
// Echoes back all received WebSocket messages
void
do_session(ws_type& ws, net::yield_context yield)
do_session(
websocket::stream<beast::tcp_stream>& ws,
net::yield_context yield)
{
beast::error_code ec;
@@ -133,10 +130,11 @@ do_listen(
fail(ec, "accept");
else
net::spawn(
acceptor.get_executor().context(),
acceptor.get_executor(),
std::bind(
&do_session,
ws_type(std::move(socket)),
websocket::stream<
beast::tcp_stream>(std::move(socket)),
std::placeholders::_1));
}
}
@@ -157,7 +155,7 @@ int main(int argc, char* argv[])
auto const threads = std::max<int>(1, std::atoi(argv[3]));
// The io_context is required for all I/O
net::io_context ioc{threads};
net::io_context ioc(threads);
// Spawn a listening port
net::spawn(ioc,
@@ -78,13 +78,8 @@ setup_stream(websocket::stream<NextLayer>& ws)
//------------------------------------------------------------------------------
// The type of websocket stream to use
// Stackful coroutines are already stranded.
using ws_type = websocket::stream<
beast::tcp_stream<net::io_context::executor_type>>;
void
do_sync_session(ws_type& ws)
do_sync_session(websocket::stream<beast::tcp_stream>& ws)
{
beast::error_code ec;
@@ -135,7 +130,8 @@ do_sync_listen(
std::thread(std::bind(
&do_sync_session,
ws_type(std::move(socket)))).detach();
websocket::stream<beast::tcp_stream>(
std::move(socket)))).detach();
}
}
@@ -144,13 +140,13 @@ do_sync_listen(
// Echoes back all received WebSocket messages
class async_session : public std::enable_shared_from_this<async_session>
{
websocket::stream<beast::tcp_stream<net::io_context::strand>> ws_;
websocket::stream<beast::tcp_stream> ws_;
beast::multi_buffer buffer_;
public:
// Take ownership of the socket
explicit
async_session(tcp::socket socket)
async_session(tcp::socket&& socket)
: ws_(std::move(socket))
{
setup_stream(ws_);
@@ -160,6 +156,11 @@ public:
void
run()
{
// Set suggested timeout settings for the websocket
ws_.set_option(
websocket::stream_base::suggested_settings(
websocket::role_type::server));
// Set a decorator to change the Server of the handshake
ws_.set_option(websocket::stream_base::decorator(
[](websocket::response_type& res)
@@ -240,13 +241,17 @@ public:
// Accepts incoming connections and launches the sessions
class async_listener : public std::enable_shared_from_this<async_listener>
{
net::io_context& ioc_;
tcp::acceptor acceptor_;
tcp::socket socket_;
public:
async_listener(
net::io_context& ioc,
tcp::endpoint endpoint)
: acceptor_(ioc)
: ioc_(ioc)
, acceptor_(beast::make_strand(ioc))
, socket_(beast::make_strand(ioc))
{
beast::error_code ec;
@@ -297,13 +302,14 @@ public:
do_accept()
{
acceptor_.async_accept(
socket_,
beast::bind_front_handler(
&async_listener::on_accept,
shared_from_this()));
}
void
on_accept(beast::error_code ec, tcp::socket socket)
on_accept(beast::error_code ec)
{
if(ec)
{
@@ -312,9 +318,12 @@ public:
else
{
// Create the async_session and run it
std::make_shared<async_session>(std::move(socket))->run();
std::make_shared<async_session>(std::move(socket_))->run();
}
// Make sure each session gets its own strand
socket_ = tcp::socket(beast::make_strand(ioc_));
// Accept another connection
do_accept();
}
@@ -323,12 +332,19 @@ public:
//------------------------------------------------------------------------------
void
do_coro_session(ws_type& ws, net::yield_context yield)
do_coro_session(
websocket::stream<beast::tcp_stream>& ws,
net::yield_context yield)
{
beast::error_code ec;
setup_stream(ws);
// Set suggested timeout settings for the websocket
ws.set_option(
websocket::stream_base::suggested_settings(
websocket::role_type::server));
// Set a decorator to change the Server of the handshake
ws.set_option(websocket::stream_base::decorator(
[](websocket::response_type& res)
@@ -395,10 +411,11 @@ do_coro_listen(
}
net::spawn(
acceptor.get_executor().context(),
acceptor.get_executor(),
std::bind(
&do_coro_session,
ws_type(std::move(socket)),
websocket::stream<
beast::tcp_stream>(std::move(socket)),
std::placeholders::_1));
}
}
@@ -51,8 +51,7 @@ class session
: public net::coroutine
, public std::enable_shared_from_this<session>
{
websocket::stream<beast::ssl_stream<
beast::tcp_stream<net::io_context::strand>>> ws_;
websocket::stream<beast::ssl_stream<beast::tcp_stream>> ws_;
beast::multi_buffer buffer_;
public:
@@ -166,6 +165,7 @@ class listener
: public net::coroutine
, public std::enable_shared_from_this<listener>
{
net::io_context& ioc_;
ssl::context& ctx_;
tcp::acceptor acceptor_;
tcp::socket socket_;
@@ -175,7 +175,8 @@ public:
net::io_context& ioc,
ssl::context& ctx,
tcp::endpoint endpoint)
: ctx_(ctx)
: ioc_(ioc)
, ctx_(ctx)
, acceptor_(ioc)
, socket_(ioc)
{
@@ -247,6 +248,9 @@ public:
// Create the session and run it
std::make_shared<session>(std::move(socket_), ctx_)->run();
}
// Make sure each session gets its own strand
socket_ = tcp::socket(beast::make_strand(ioc_));
}
}
}
@@ -46,8 +46,7 @@ class session
: public net::coroutine
, public std::enable_shared_from_this<session>
{
websocket::stream<
beast::tcp_stream<net::io_context::strand>> ws_;
websocket::stream<beast::tcp_stream> ws_;
beast::multi_buffer buffer_;
public:
@@ -143,6 +142,7 @@ class listener
: public net::coroutine
, public std::enable_shared_from_this<listener>
{
net::io_context& ioc_;
tcp::acceptor acceptor_;
tcp::socket socket_;
@@ -150,8 +150,9 @@ public:
listener(
net::io_context& ioc,
tcp::endpoint endpoint)
: acceptor_(ioc)
, socket_(ioc)
: ioc_(ioc)
, acceptor_(beast::make_strand(ioc))
, socket_(beast::make_strand(ioc))
{
beast::error_code ec;
@@ -221,6 +222,9 @@ public:
// Create the session and run it
std::make_shared<session>(std::move(socket_))->run();
}
// Make sure each session gets its own strand
socket_ = tcp::socket(beast::make_strand(ioc_));
}
}
}