boostorg/beast
1
0
Fork 1
mirror of https://github.com/boostorg/beast.git synced 2025-07-31 05:17:26 +02:00
Code Issues Packages Projects Releases Wiki Activity

Use tcp_stream, HTTP timeouts in advanced servers

Browse Source
This commit is contained in:
Vinnie Falco
2019-02-10 05:48:29 -08:00
parent 6ee1a88292
commit 45d722874f
4 changed files with 236 additions and 387 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGELOG.md
View File

@ -6,6 +6,7 @@ Version 213:
* Use timeouts in HTTP client examples * Use timeouts in HTTP client examples
* Use tcp_stream in WebSocket client examples * Use tcp_stream in WebSocket client examples
* Use tcp_stream in WebSocket server examples * Use tcp_stream in WebSocket server examples
* Use tcp_stream, HTTP timeouts in advanced servers
-------------------------------------------------------------------------------- --------------------------------------------------------------------------------

448
example/advanced/server-flex/advanced_server_flex.cpp
View File

@ -22,11 +22,9 @@
#include <boost/beast/version.hpp> #include <boost/beast/version.hpp>
#include <boost/beast/_experimental/core/ssl_stream.hpp> #include <boost/beast/_experimental/core/ssl_stream.hpp>
#include <boost/asio/bind_executor.hpp> #include <boost/asio/bind_executor.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/signal_set.hpp> #include <boost/asio/signal_set.hpp>
#include <boost/asio/ssl/stream.hpp>
#include <boost/asio/strand.hpp>
#include <boost/asio/steady_timer.hpp> #include <boost/asio/steady_timer.hpp>
#include <boost/asio/strand.hpp>
#include <boost/make_unique.hpp> #include <boost/make_unique.hpp>
#include <boost/config.hpp> #include <boost/config.hpp>
#include <algorithm> #include <algorithm>
@ -244,16 +242,13 @@ class websocket_session
char ping_state_ = 0; char ping_state_ = 0;
protected: protected:
net::strand<
net::io_context::executor_type> strand_;
net::steady_timer timer_; net::steady_timer timer_;
public: public:
// Construct the session // Construct the session
explicit explicit
websocket_session(net::io_context& ioc) websocket_session(net::io_context& ioc)
: strand_(ioc.get_executor()) : timer_(ioc,
, timer_(ioc,
(std::chrono::steady_clock::time_point::max)()) (std::chrono::steady_clock::time_point::max)())
{ {
} }
@ -272,18 +267,15 @@ public:
std::placeholders::_1, std::placeholders::_1,
std::placeholders::_2)); std::placeholders::_2));
// Set the timer // VFALCO What about the timer?
timer_.expires_after(std::chrono::seconds(15));
// Accept the websocket handshake // Accept the websocket handshake
derived().ws().async_accept( derived().ws().async_accept(
req, req,
net::bind_executor( std::bind(
strand_, &websocket_session::on_accept,
std::bind( derived().shared_from_this(),
&websocket_session::on_accept, std::placeholders::_1));
derived().shared_from_this(),
std::placeholders::_1)));
} }
void void
@ -322,12 +314,10 @@ public:
// Now send the ping // Now send the ping
derived().ws().async_ping({}, derived().ws().async_ping({},
net::bind_executor( std::bind(
strand_, &websocket_session::on_ping,
std::bind( derived().shared_from_this(),
&websocket_session::on_ping, std::placeholders::_1));
derived().shared_from_this(),
std::placeholders::_1)));
} }
else else
{ {
@ -343,7 +333,7 @@ public:
// Wait on the timer // Wait on the timer
timer_.async_wait( timer_.async_wait(
net::bind_executor( net::bind_executor(
strand_, derived().ws().get_executor(), // use the strand
std::bind( std::bind(
&websocket_session::on_timer, &websocket_session::on_timer,
derived().shared_from_this(), derived().shared_from_this(),
@ -403,13 +393,11 @@ public:
// Read a message into our buffer // Read a message into our buffer
derived().ws().async_read( derived().ws().async_read(
buffer_, buffer_,
net::bind_executor( std::bind(
strand_, &websocket_session::on_read,
std::bind( derived().shared_from_this(),
&websocket_session::on_read, std::placeholders::_1,
derived().shared_from_this(), std::placeholders::_2));
std::placeholders::_1,
std::placeholders::_2)));
} }
void void
@ -437,13 +425,11 @@ public:
derived().ws().text(derived().ws().got_text()); derived().ws().text(derived().ws().got_text());
derived().ws().async_write( derived().ws().async_write(
buffer_.data(), buffer_.data(),
net::bind_executor( std::bind(
strand_, &websocket_session::on_write,
std::bind( derived().shared_from_this(),
&websocket_session::on_write, std::placeholders::_1,
derived().shared_from_this(), std::placeholders::_2));
std::placeholders::_1,
std::placeholders::_2)));
} }
void void
@ -473,21 +459,24 @@ class plain_websocket_session
: public websocket_session<plain_websocket_session> : public websocket_session<plain_websocket_session>
, public std::enable_shared_from_this<plain_websocket_session> , public std::enable_shared_from_this<plain_websocket_session>
{ {
websocket::stream<tcp::socket> ws_; websocket::stream<
beast::tcp_stream<net::io_context::strand>> ws_;
bool close_ = false; bool close_ = false;
public: public:
// Create the session // Create the session
explicit explicit
plain_websocket_session(tcp::socket socket) plain_websocket_session(
beast::tcp_stream<net::io_context::strand>&& stream)
: websocket_session<plain_websocket_session>( : websocket_session<plain_websocket_session>(
socket.get_executor().context()) stream.get_executor().context())
, ws_(std::move(socket)) , ws_(std::move(stream))
{ {
} }
// Called by the base class // Called by the base class
websocket::stream<tcp::socket>& websocket::stream<
beast::tcp_stream<net::io_context::strand>>&
ws() ws()
{ {
return ws_; return ws_;
@ -514,18 +503,17 @@ public:
return; return;
close_ = true; close_ = true;
// VFALCO This doesn't look right...
// Set the timer // Set the timer
timer_.expires_after(std::chrono::seconds(15)); timer_.expires_after(std::chrono::seconds(15));
// Close the WebSocket Connection // Close the WebSocket Connection
ws_.async_close( ws_.async_close(
websocket::close_code::normal, websocket::close_code::normal,
net::bind_executor( std::bind(
strand_, &plain_websocket_session::on_close,
std::bind( shared_from_this(),
&plain_websocket_session::on_close, std::placeholders::_1));
shared_from_this(),
std::placeholders::_1)));
} }
void void
@ -547,13 +535,15 @@ class ssl_websocket_session
: public websocket_session<ssl_websocket_session> : public websocket_session<ssl_websocket_session>
, public std::enable_shared_from_this<ssl_websocket_session> , public std::enable_shared_from_this<ssl_websocket_session>
{ {
websocket::stream<beast::ssl_stream<tcp::socket>> ws_; websocket::stream<beast::ssl_stream<
beast::tcp_stream<net::io_context::strand>>> ws_;
bool eof_ = false; bool eof_ = false;
public: public:
// Create the http_session // Create the http_session
explicit explicit
ssl_websocket_session(beast::ssl_stream<tcp::socket> stream) ssl_websocket_session(beast::ssl_stream<
beast::tcp_stream<net::io_context::strand>>&& stream)
: websocket_session<ssl_websocket_session>( : websocket_session<ssl_websocket_session>(
stream.get_executor().context()) stream.get_executor().context())
, ws_(std::move(stream)) , ws_(std::move(stream))
@ -561,7 +551,8 @@ public:
} }
// Called by the base class // Called by the base class
websocket::stream<beast::ssl_stream<tcp::socket>>& websocket::stream<beast::ssl_stream<
beast::tcp_stream<net::io_context::strand>>>&
ws() ws()
{ {
return ws_; return ws_;
@ -590,12 +581,10 @@ public:
// Perform the SSL shutdown // Perform the SSL shutdown
ws_.next_layer().async_shutdown( ws_.next_layer().async_shutdown(
net::bind_executor( std::bind(
strand_, &ssl_websocket_session::on_shutdown,
std::bind( shared_from_this(),
&ssl_websocket_session::on_shutdown, std::placeholders::_1));
shared_from_this(),
std::placeholders::_1)));
} }
void void
@ -629,17 +618,17 @@ public:
template<class Body, class Allocator> template<class Body, class Allocator>
void void
make_websocket_session( make_websocket_session(
tcp::socket socket, beast::tcp_stream<net::io_context::strand> stream,
http::request<Body, http::basic_fields<Allocator>> req) http::request<Body, http::basic_fields<Allocator>> req)
{ {
std::make_shared<plain_websocket_session>( std::make_shared<plain_websocket_session>(
std::move(socket))->run(std::move(req)); std::move(stream))->run(std::move(req));
} }
template<class Body, class Allocator> template<class Body, class Allocator>
void void
make_websocket_session( make_websocket_session(
beast::ssl_stream<tcp::socket> stream, beast::ssl_stream<beast::tcp_stream<net::io_context::strand>> stream,
http::request<Body, http::basic_fields<Allocator>> req) http::request<Body, http::basic_fields<Allocator>> req)
{ {
std::make_shared<ssl_websocket_session>( std::make_shared<ssl_websocket_session>(
@ -735,13 +724,11 @@ class http_session
http::async_write( http::async_write(
self_.derived().stream(), self_.derived().stream(),
msg_, msg_,
net::bind_executor( std::bind(
self_.strand_, &http_session::on_write,
std::bind( self_.derived().shared_from_this(),
&http_session::on_write, std::placeholders::_1,
self_.derived().shared_from_this(), msg_.need_eof()));
std::placeholders::_1,
msg_.need_eof())));
} }
}; };
@ -761,8 +748,6 @@ class http_session
protected: protected:
net::steady_timer timer_; net::steady_timer timer_;
net::strand<
net::io_context::executor_type> strand_;
beast::flat_buffer buffer_; beast::flat_buffer buffer_;
public: public:
@ -775,7 +760,6 @@ public:
, queue_(*this) , queue_(*this)
, timer_(ioc, , timer_(ioc,
(std::chrono::steady_clock::time_point::max)()) (std::chrono::steady_clock::time_point::max)())
, strand_(ioc.get_executor())
, buffer_(std::move(buffer)) , buffer_(std::move(buffer))
{ {
} }
@ -783,58 +767,28 @@ public:
void void
do_read() do_read()
{ {
// Set the timer
timer_.expires_after(std::chrono::seconds(15));
// Make the request empty before reading, // Make the request empty before reading,
// otherwise the operation behavior is undefined. // otherwise the operation behavior is undefined.
req_ = {}; req_ = {};
// Set the timeout.
beast::get_lowest_layer(
derived().stream()).expires_after(std::chrono::seconds(30));
// Read a request // Read a request
http::async_read( http::async_read(
derived().stream(), derived().stream(),
buffer_, buffer_,
req_, req_,
net::bind_executor( std::bind(
strand_, &http_session::on_read,
std::bind( derived().shared_from_this(),
&http_session::on_read, std::placeholders::_1));
derived().shared_from_this(),
std::placeholders::_1)));
}
// Called when the timer expires.
void
on_timer(beast::error_code ec)
{
if(ec && ec != net::error::operation_aborted)
return fail(ec, "timer");
// Check if this has been upgraded to Websocket
if(timer_.expiry() == (std::chrono::steady_clock::time_point::min)())
return;
// Verify that the timer really expired since the deadline may have moved.
if(timer_.expiry() <= std::chrono::steady_clock::now())
return derived().do_timeout();
// Wait on the timer
timer_.async_wait(
net::bind_executor(
strand_,
std::bind(
&http_session::on_timer,
derived().shared_from_this(),
std::placeholders::_1)));
} }
void void
on_read(beast::error_code ec) on_read(beast::error_code ec)
{ {
// Happens when the timer closes the socket
if(ec == net::error::operation_aborted)
return;
// This means they closed the connection // This means they closed the connection
if(ec == http::error::end_of_stream) if(ec == http::error::end_of_stream)
return derived().do_eof(); return derived().do_eof();
@ -845,10 +799,6 @@ public:
// See if it is a WebSocket Upgrade // See if it is a WebSocket Upgrade
if(websocket::is_upgrade(req_)) if(websocket::is_upgrade(req_))
{ {
// Make timer expire immediately, by setting expiry to time_point::min we can detect
// the upgrade to websocket in the timer handler
timer_.expires_at((std::chrono::steady_clock::time_point::min)());
// Transfer the stream to a new WebSocket session // Transfer the stream to a new WebSocket session
return make_websocket_session( return make_websocket_session(
derived().release_stream(), derived().release_stream(),
@ -894,115 +844,31 @@ class plain_http_session
: public http_session<plain_http_session> : public http_session<plain_http_session>
, public std::enable_shared_from_this<plain_http_session> , public std::enable_shared_from_this<plain_http_session>
{ {
tcp::socket socket_; beast::tcp_stream<net::io_context::strand> stream_;
net::strand<
net::io_context::executor_type> strand_;
public: public:
// Create the http_session // Create the http_session
plain_http_session( plain_http_session(
tcp::socket socket, beast::tcp_stream<net::io_context::strand>&& stream,
beast::flat_buffer buffer, beast::flat_buffer&& buffer,
std::shared_ptr<std::string const> const& doc_root) std::shared_ptr<std::string const> const& doc_root)
: http_session<plain_http_session>( : http_session<plain_http_session>(
socket.get_executor().context(), stream.get_executor().context(),
std::move(buffer), std::move(buffer),
doc_root) doc_root)
, socket_(std::move(socket)) , stream_(std::move(stream))
, strand_(socket_.get_executor())
{ {
} }
// Called by the base class // Called by the base class
tcp::socket& beast::tcp_stream<net::io_context::strand>&
stream()
{
return socket_;
}
// Called by the base class
tcp::socket
release_stream()
{
return std::move(socket_);
}
// Start the asynchronous operation
void
run()
{
// Make sure we run on the strand
if(! strand_.running_in_this_thread())
return net::post(
net::bind_executor(
strand_,
std::bind(
&plain_http_session::run,
shared_from_this())));
// Run the timer. The timer is operated
// continuously, this simplifies the code.
on_timer({});
do_read();
}
void
do_eof()
{
// Send a TCP shutdown
beast::error_code ec;
socket_.shutdown(tcp::socket::shutdown_send, ec);
// At this point the connection is closed gracefully
}
void
do_timeout()
{
// Closing the socket cancels all outstanding operations. They
// will complete with net::error::operation_aborted
beast::error_code ec;
socket_.shutdown(tcp::socket::shutdown_both, ec);
socket_.close(ec);
}
};
// Handles an SSL HTTP connection
class ssl_http_session
: public http_session<ssl_http_session>
, public std::enable_shared_from_this<ssl_http_session>
{
beast::ssl_stream<tcp::socket> stream_;
net::strand<
net::io_context::executor_type> strand_;
bool eof_ = false;
public:
// Create the http_session
ssl_http_session(
tcp::socket socket,
ssl::context& ctx,
beast::flat_buffer buffer,
std::shared_ptr<std::string const> const& doc_root)
: http_session<ssl_http_session>(
socket.get_executor().context(),
std::move(buffer),
doc_root)
, stream_(std::move(socket), ctx)
, strand_(stream_.get_executor())
{
}
// Called by the base class
beast::ssl_stream<tcp::socket>&
stream() stream()
{ {
return stream_; return stream_;
} }
// Called by the base class // Called by the base class
beast::ssl_stream<tcp::socket> beast::tcp_stream<net::io_context::strand>
release_stream() release_stream()
{ {
return std::move(stream_); return std::move(stream_);
@ -1013,43 +879,108 @@ public:
run() run()
{ {
// Make sure we run on the strand // Make sure we run on the strand
if(! strand_.running_in_this_thread()) if(! stream_.get_executor().running_in_this_thread())
return net::post( return net::post(
net::bind_executor( stream_.get_executor(),
strand_, std::bind(
std::bind( &plain_http_session::run,
&ssl_http_session::run, shared_from_this()));
shared_from_this())));
// Run the timer. The timer is operated do_read();
// continuously, this simplifies the code. }
on_timer({});
// Set the timer void
timer_.expires_after(std::chrono::seconds(15)); do_eof()
{
// Send a TCP shutdown
beast::error_code ec;
stream_.socket().shutdown(tcp::socket::shutdown_send, ec);
// At this point the connection is closed gracefully
}
void
do_timeout()
{
// Closing the socket cancels all outstanding operations. They
// will complete with net::error::operation_aborted
beast::error_code ec;
stream_.socket().shutdown(tcp::socket::shutdown_both, ec);
stream_.socket().close(ec);
}
};
// Handles an SSL HTTP connection
class ssl_http_session
: public http_session<ssl_http_session>
, public std::enable_shared_from_this<ssl_http_session>
{
beast::ssl_stream<
beast::tcp_stream<net::io_context::strand>> stream_;
bool eof_ = false;
public:
// Create the http_session
ssl_http_session(
beast::tcp_stream<net::io_context::strand>&& stream,
ssl::context& ctx,
beast::flat_buffer&& buffer,
std::shared_ptr<std::string const> const& doc_root)
: http_session<ssl_http_session>(
stream.get_executor().context(),
std::move(buffer),
doc_root)
, stream_(std::move(stream), ctx)
{
}
// Called by the base class
beast::ssl_stream<
beast::tcp_stream<net::io_context::strand>>&
stream()
{
return stream_;
}
// Called by the base class
beast::ssl_stream<
beast::tcp_stream<net::io_context::strand>>
release_stream()
{
return std::move(stream_);
}
// Start the asynchronous operation
void
run()
{
// Make sure we run on the strand
if(! stream_.get_executor().running_in_this_thread())
return net::post(
stream_.get_executor(),
std::bind(
&ssl_http_session::run,
shared_from_this()));
// Set the timeout.
beast::get_lowest_layer(stream_).expires_after(std::chrono::seconds(30));
// Perform the SSL handshake // Perform the SSL handshake
// Note, this is the buffered version of the handshake. // Note, this is the buffered version of the handshake.
stream_.async_handshake( stream_.async_handshake(
ssl::stream_base::server, ssl::stream_base::server,
buffer_.data(), buffer_.data(),
net::bind_executor( std::bind(
strand_, &ssl_http_session::on_handshake,
std::bind( shared_from_this(),
&ssl_http_session::on_handshake, std::placeholders::_1,
shared_from_this(), std::placeholders::_2));
std::placeholders::_1,
std::placeholders::_2)));
} }
void void
on_handshake( on_handshake(
beast::error_code ec, beast::error_code ec,
std::size_t bytes_used) std::size_t bytes_used)
{ {
// Happens when the handshake times out
if(ec == net::error::operation_aborted)
return;
if(ec) if(ec)
return fail(ec, "handshake"); return fail(ec, "handshake");
@ -1064,17 +995,15 @@ public:
{ {
eof_ = true; eof_ = true;
// Set the timer // Set the timeout.
timer_.expires_after(std::chrono::seconds(15)); beast::get_lowest_layer(stream_).expires_after(std::chrono::seconds(30));
// Perform the SSL shutdown // Perform the SSL shutdown
stream_.async_shutdown( stream_.async_shutdown(
net::bind_executor( std::bind(
strand_, &ssl_http_session::on_shutdown,
std::bind( shared_from_this(),
&ssl_http_session::on_shutdown, std::placeholders::_1));
shared_from_this(),
std::placeholders::_1)));
} }
void void
@ -1089,20 +1018,6 @@ public:
// At this point the connection is closed gracefully // At this point the connection is closed gracefully
} }
void
do_timeout()
{
// If this is true it means we timed out performing the shutdown
if(eof_)
return;
// Start the timer again
timer_.expires_at(
(std::chrono::steady_clock::time_point::max)());
on_timer({});
do_eof();
}
}; };
//------------------------------------------------------------------------------ //------------------------------------------------------------------------------
@ -1110,10 +1025,8 @@ public:
// Detects SSL handshakes // Detects SSL handshakes
class detect_session : public std::enable_shared_from_this<detect_session> class detect_session : public std::enable_shared_from_this<detect_session>
{ {
tcp::socket socket_; beast::tcp_stream<net::io_context::strand> stream_;
ssl::context& ctx_; ssl::context& ctx_;
net::strand<
net::io_context::executor_type> strand_;
std::shared_ptr<std::string const> doc_root_; std::shared_ptr<std::string const> doc_root_;
beast::flat_buffer buffer_; beast::flat_buffer buffer_;
@ -1123,9 +1036,8 @@ public:
tcp::socket socket, tcp::socket socket,
ssl::context& ctx, ssl::context& ctx,
std::shared_ptr<std::string const> const& doc_root) std::shared_ptr<std::string const> const& doc_root)
: socket_(std::move(socket)) : stream_(std::move(socket))
, ctx_(ctx) , ctx_(ctx)
, strand_(socket_.get_executor())
, doc_root_(doc_root) , doc_root_(doc_root)
{ {
} }
@ -1134,17 +1046,17 @@ public:
void void
run() run()
{ {
async_detect_ssl( // Set the timeout.
socket_, stream_.expires_after(std::chrono::seconds(30));
buffer_,
net::bind_executor(
strand_,
std::bind(
&detect_session::on_detect,
shared_from_this(),
std::placeholders::_1,
std::placeholders::_2)));
async_detect_ssl(
stream_,
buffer_,
std::bind(
&detect_session::on_detect,
shared_from_this(),
std::placeholders::_1,
std::placeholders::_2));
} }
void void
@ -1157,7 +1069,7 @@ public:
{ {
// Launch SSL session // Launch SSL session
std::make_shared<ssl_http_session>( std::make_shared<ssl_http_session>(
std::move(socket_), std::move(stream_),
ctx_, ctx_,
std::move(buffer_), std::move(buffer_),
doc_root_)->run(); doc_root_)->run();
@ -1166,7 +1078,7 @@ public:
// Launch plain session // Launch plain session
std::make_shared<plain_http_session>( std::make_shared<plain_http_session>(
std::move(socket_), std::move(stream_),
std::move(buffer_), std::move(buffer_),
doc_root_)->run(); doc_root_)->run();
} }
@ -1177,7 +1089,6 @@ class listener : public std::enable_shared_from_this<listener>
{ {
ssl::context& ctx_; ssl::context& ctx_;
tcp::acceptor acceptor_; tcp::acceptor acceptor_;
tcp::socket socket_;
std::shared_ptr<std::string const> doc_root_; std::shared_ptr<std::string const> doc_root_;
public: public:
@ -1188,7 +1099,6 @@ public:
std::shared_ptr<std::string const> const& doc_root) std::shared_ptr<std::string const> const& doc_root)
: ctx_(ctx) : ctx_(ctx)
, acceptor_(ioc) , acceptor_(ioc)
, socket_(ioc)
, doc_root_(doc_root) , doc_root_(doc_root)
{ {
beast::error_code ec; beast::error_code ec;
@ -1240,15 +1150,15 @@ public:
do_accept() do_accept()
{ {
acceptor_.async_accept( acceptor_.async_accept(
socket_,
std::bind( std::bind(
&listener::on_accept, &listener::on_accept,
shared_from_this(), shared_from_this(),
std::placeholders::_1)); std::placeholders::_1,
std::placeholders::_2));
} }
void void
on_accept(beast::error_code ec) on_accept(beast::error_code ec, tcp::socket socket)
{ {
if(ec) if(ec)
{ {
@ -1258,7 +1168,7 @@ public:
{ {
// Create the detector http_session and run it // Create the detector http_session and run it
std::make_shared<detect_session>( std::make_shared<detect_session>(
std::move(socket_), std::move(socket),
ctx_, ctx_,
doc_root_)->run(); doc_root_)->run();
} }

171
example/advanced/server/advanced_server.cpp
View File

@ -18,10 +18,9 @@
#include <boost/beast/websocket.hpp> #include <boost/beast/websocket.hpp>
#include <boost/beast/version.hpp> #include <boost/beast/version.hpp>
#include <boost/asio/bind_executor.hpp> #include <boost/asio/bind_executor.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/signal_set.hpp> #include <boost/asio/signal_set.hpp>
#include <boost/asio/strand.hpp>
#include <boost/asio/steady_timer.hpp> #include <boost/asio/steady_timer.hpp>
#include <boost/asio/strand.hpp>
#include <boost/make_unique.hpp> #include <boost/make_unique.hpp>
#include <boost/config.hpp> #include <boost/config.hpp>
#include <algorithm> #include <algorithm>
@ -221,9 +220,7 @@ fail(beast::error_code ec, char const* what)
// Echoes back all received WebSocket messages // Echoes back all received WebSocket messages
class websocket_session : public std::enable_shared_from_this<websocket_session> class websocket_session : public std::enable_shared_from_this<websocket_session>
{ {
websocket::stream<tcp::socket> ws_; websocket::stream<beast::tcp_stream<net::io_context::strand>> ws_;
net::strand<
net::io_context::executor_type> strand_;
net::steady_timer timer_; net::steady_timer timer_;
beast::flat_buffer buffer_; beast::flat_buffer buffer_;
char ping_state_ = 0; char ping_state_ = 0;
@ -233,13 +230,12 @@ public:
explicit explicit
websocket_session(tcp::socket socket) websocket_session(tcp::socket socket)
: ws_(std::move(socket)) : ws_(std::move(socket))
, strand_(ws_.get_executor())
, timer_(ws_.get_executor().context(), , timer_(ws_.get_executor().context(),
(std::chrono::steady_clock::time_point::max)()) (std::chrono::steady_clock::time_point::max)())
{ {
} }
// Start the asynchronous operation // Start the asynchronous accept operation
template<class Body, class Allocator> template<class Body, class Allocator>
void void
do_accept(http::request<Body, http::basic_fields<Allocator>> req) do_accept(http::request<Body, http::basic_fields<Allocator>> req)
@ -263,12 +259,10 @@ public:
// Accept the websocket handshake // Accept the websocket handshake
ws_.async_accept( ws_.async_accept(
req, req,
net::bind_executor( std::bind(
strand_, &websocket_session::on_accept,
std::bind( shared_from_this(),
&websocket_session::on_accept, std::placeholders::_1));
shared_from_this(),
std::placeholders::_1)));
} }
void void
@ -307,12 +301,10 @@ public:
// Now send the ping // Now send the ping
ws_.async_ping({}, ws_.async_ping({},
net::bind_executor( std::bind(
strand_, &websocket_session::on_ping,
std::bind( shared_from_this(),
&websocket_session::on_ping, std::placeholders::_1));
shared_from_this(),
std::placeholders::_1)));
} }
else else
{ {
@ -322,8 +314,8 @@ public:
// Closing the socket cancels all outstanding operations. They // Closing the socket cancels all outstanding operations. They
// will complete with net::error::operation_aborted // will complete with net::error::operation_aborted
ws_.next_layer().shutdown(tcp::socket::shutdown_both, ec); beast::get_lowest_layer(ws_).socket().shutdown(tcp::socket::shutdown_both, ec);
ws_.next_layer().close(ec); beast::get_lowest_layer(ws_).socket().close(ec);
return; return;
} }
} }
@ -331,7 +323,7 @@ public:
// Wait on the timer // Wait on the timer
timer_.async_wait( timer_.async_wait(
net::bind_executor( net::bind_executor(
strand_, ws_.get_executor(), // use the strand
std::bind( std::bind(
&websocket_session::on_timer, &websocket_session::on_timer,
shared_from_this(), shared_from_this(),
@ -391,13 +383,11 @@ public:
// Read a message into our buffer // Read a message into our buffer
ws_.async_read( ws_.async_read(
buffer_, buffer_,
net::bind_executor( std::bind(
strand_, &websocket_session::on_read,
std::bind( shared_from_this(),
&websocket_session::on_read, std::placeholders::_1,
shared_from_this(), std::placeholders::_2));
std::placeholders::_1,
std::placeholders::_2)));
} }
void void
@ -425,13 +415,11 @@ public:
ws_.text(ws_.got_text()); ws_.text(ws_.got_text());
ws_.async_write( ws_.async_write(
buffer_.data(), buffer_.data(),
net::bind_executor( std::bind(
strand_, &websocket_session::on_write,
std::bind( shared_from_this(),
&websocket_session::on_write, std::placeholders::_1,
shared_from_this(), std::placeholders::_2));
std::placeholders::_1,
std::placeholders::_2)));
} }
void void
@ -530,15 +518,13 @@ class http_session : public std::enable_shared_from_this<http_session>
operator()() operator()()
{ {
http::async_write( http::async_write(
self_.socket_, self_.stream_,
msg_, msg_,
net::bind_executor( std::bind(
self_.strand_, &http_session::on_write,
std::bind( self_.shared_from_this(),
&http_session::on_write, std::placeholders::_1,
self_.shared_from_this(), msg_.need_eof()));
std::placeholders::_1,
msg_.need_eof())));
} }
}; };
@ -552,10 +538,7 @@ class http_session : public std::enable_shared_from_this<http_session>
} }
}; };
tcp::socket socket_; beast::tcp_stream<net::io_context::strand> stream_;
net::strand<
net::io_context::executor_type> strand_;
net::steady_timer timer_;
beast::flat_buffer buffer_; beast::flat_buffer buffer_;
std::shared_ptr<std::string const> doc_root_; std::shared_ptr<std::string const> doc_root_;
http::request<http::string_body> req_; http::request<http::string_body> req_;
@ -567,10 +550,7 @@ public:
http_session( http_session(
tcp::socket socket, tcp::socket socket,
std::shared_ptr<std::string const> const& doc_root) std::shared_ptr<std::string const> const& doc_root)
: socket_(std::move(socket)) : stream_(std::move(socket))
, strand_(socket_.get_executor())
, timer_(socket_.get_executor().context(),
(std::chrono::steady_clock::time_point::max)())
, doc_root_(doc_root) , doc_root_(doc_root)
, queue_(*this) , queue_(*this)
{ {
@ -581,17 +561,12 @@ public:
run() run()
{ {
// Make sure we run on the strand // Make sure we run on the strand
if(! strand_.running_in_this_thread()) if(! stream_.get_executor().running_in_this_thread())
return net::post( return net::post(
net::bind_executor( stream_.get_executor(),
strand_, std::bind(
std::bind( &http_session::run,
&http_session::run, shared_from_this()));
shared_from_this())));
// Run the timer. The timer is operated
// continuously, this simplifies the code.
on_timer({});
do_read(); do_read();
} }
@ -599,61 +574,25 @@ public:
void void
do_read() do_read()
{ {
// Set the timer
timer_.expires_after(std::chrono::seconds(15));
// Make the request empty before reading, // Make the request empty before reading,
// otherwise the operation behavior is undefined. // otherwise the operation behavior is undefined.
req_ = {}; req_ = {};
// Set the timeout.
stream_.expires_after(std::chrono::seconds(30));
// Read a request // Read a request
http::async_read(socket_, buffer_, req_, http::async_read(stream_, buffer_, req_,
net::bind_executor( std::bind(
strand_, &http_session::on_read,
std::bind( shared_from_this(),
&http_session::on_read, std::placeholders::_1));
shared_from_this(),
std::placeholders::_1)));
}
// Called when the timer expires.
void
on_timer(beast::error_code ec)
{
if(ec && ec != net::error::operation_aborted)
return fail(ec, "timer");
// Check if this has been upgraded to Websocket
if(timer_.expiry() == (std::chrono::steady_clock::time_point::min)())
return;
// Verify that the timer really expired since the deadline may have moved.
if(timer_.expiry() <= std::chrono::steady_clock::now())
{
// Closing the socket cancels all outstanding operations. They
// will complete with net::error::operation_aborted
socket_.shutdown(tcp::socket::shutdown_both, ec);
socket_.close(ec);
return;
}
// Wait on the timer
timer_.async_wait(
net::bind_executor(
strand_,
std::bind(
&http_session::on_timer,
shared_from_this(),
std::placeholders::_1)));
} }
void void
on_read(beast::error_code ec) on_read(beast::error_code ec)
{ {
// Happens when the timer closes the socket
if(ec == net::error::operation_aborted)
return;
// This means they closed the connection // This means they closed the connection
if(ec == http::error::end_of_stream) if(ec == http::error::end_of_stream)
return do_close(); return do_close();
@ -664,13 +603,9 @@ public:
// See if it is a WebSocket Upgrade // See if it is a WebSocket Upgrade
if(websocket::is_upgrade(req_)) if(websocket::is_upgrade(req_))
{ {
// Make timer expire immediately, by setting expiry to time_point::min we can detect // Create a websocket session by transferring the socket
// the upgrade to websocket in the timer handler
timer_.expires_at((std::chrono::steady_clock::time_point::min)());
// Create a WebSocket websocket_session by transferring the socket
std::make_shared<websocket_session>( std::make_shared<websocket_session>(
std::move(socket_))->do_accept(std::move(req_)); stream_.release_socket())->do_accept(std::move(req_));
return; return;
} }
@ -712,7 +647,7 @@ public:
{ {
// Send a TCP shutdown // Send a TCP shutdown
beast::error_code ec; beast::error_code ec;
socket_.shutdown(tcp::socket::shutdown_send, ec); stream_.socket().shutdown(tcp::socket::shutdown_send, ec);
// At this point the connection is closed gracefully // At this point the connection is closed gracefully
} }
@ -785,15 +720,15 @@ public:
do_accept() do_accept()
{ {
acceptor_.async_accept( acceptor_.async_accept(
socket_,
std::bind( std::bind(
&listener::on_accept, &listener::on_accept,
shared_from_this(), shared_from_this(),
std::placeholders::_1)); std::placeholders::_1,
std::placeholders::_2));
} }
void void
on_accept(beast::error_code ec) on_accept(beast::error_code ec, tcp::socket socket)
{ {
if(ec) if(ec)
{ {
@ -801,9 +736,9 @@ public:
} }
else else
{ {
// Create the http_session and run it // Create the http session and run it
std::make_shared<http_session>( std::make_shared<http_session>(
std::move(socket_), std::move(socket),
doc_root_)->run(); doc_root_)->run();
} }

3
include/boost/beast/core/impl/basic_stream.hpp
View File

@ -718,6 +718,9 @@ async_connect(
} }
//------------------------------------------------------------------------------ //------------------------------------------------------------------------------
//
// Customization points
//
#if ! BOOST_BEAST_DOXYGEN #if ! BOOST_BEAST_DOXYGEN