summaryrefslogtreecommitdiff
path: root/boost/beast/websocket/impl/close.ipp
diff options
context:
space:
mode:
Diffstat (limited to 'boost/beast/websocket/impl/close.ipp')
-rw-r--r--boost/beast/websocket/impl/close.ipp133
1 files changed, 60 insertions, 73 deletions
diff --git a/boost/beast/websocket/impl/close.ipp b/boost/beast/websocket/impl/close.ipp
index 0349dd1409..7b0e1ff648 100644
--- a/boost/beast/websocket/impl/close.ipp
+++ b/boost/beast/websocket/impl/close.ipp
@@ -19,6 +19,7 @@
#include <boost/asio/associated_executor.hpp>
#include <boost/asio/coroutine.hpp>
#include <boost/asio/handler_continuation_hook.hpp>
+#include <boost/asio/handler_invoke_hook.hpp>
#include <boost/asio/post.hpp>
#include <boost/throw_exception.hpp>
#include <memory>
@@ -34,25 +35,23 @@ namespace websocket {
frame. Finally it invokes the teardown operation to shut down the
underlying connection.
*/
-template<class NextLayer>
+template<class NextLayer, bool deflateSupported>
template<class Handler>
-class stream<NextLayer>::close_op
+class stream<NextLayer, deflateSupported>::close_op
: public boost::asio::coroutine
{
struct state
{
- stream<NextLayer>& ws;
+ stream<NextLayer, deflateSupported>& ws;
detail::frame_buffer fb;
error_code ev;
- token tok;
bool cont = false;
state(
- Handler&,
- stream<NextLayer>& ws_,
+ Handler const&,
+ stream<NextLayer, deflateSupported>& ws_,
close_reason const& cr)
: ws(ws_)
- , tok(ws.tok_.unique())
{
// Serialize the close frame
ws.template write_close<
@@ -63,13 +62,15 @@ class stream<NextLayer>::close_op
handler_ptr<state, Handler> d_;
public:
+ static constexpr int id = 4; // for soft_mutex
+
close_op(close_op&&) = default;
- close_op(close_op const&) = default;
+ close_op(close_op const&) = delete;
template<class DeducedHandler>
close_op(
DeducedHandler&& h,
- stream<NextLayer>& ws,
+ stream<NextLayer, deflateSupported>& ws,
close_reason const& cr)
: d_(std::forward<DeducedHandler>(h), ws, cr)
{
@@ -81,16 +82,16 @@ public:
allocator_type
get_allocator() const noexcept
{
- return boost::asio::get_associated_allocator(d_.handler());
+ return (boost::asio::get_associated_allocator)(d_.handler());
}
using executor_type = boost::asio::associated_executor_t<
- Handler, decltype(std::declval<stream<NextLayer>&>().get_executor())>;
+ Handler, decltype(std::declval<stream<NextLayer, deflateSupported>&>().get_executor())>;
executor_type
get_executor() const noexcept
{
- return boost::asio::get_associated_executor(
+ return (boost::asio::get_associated_executor)(
d_.handler(), d_->ws.get_executor());
}
@@ -107,12 +108,21 @@ public:
return op->d_->cont || asio_handler_is_continuation(
std::addressof(op->d_.handler()));
}
+
+ template<class Function>
+ friend
+ void asio_handler_invoke(Function&& f, close_op* op)
+ {
+ using boost::asio::asio_handler_invoke;
+ asio_handler_invoke(f,
+ std::addressof(op->d_.handler()));
+ }
};
-template<class NextLayer>
+template<class NextLayer, bool deflateSupported>
template<class Handler>
void
-stream<NextLayer>::
+stream<NextLayer, deflateSupported>::
close_op<Handler>::
operator()(
error_code ec,
@@ -121,16 +131,12 @@ operator()(
{
using beast::detail::clamp;
auto& d = *d_;
- close_code code{};
d.cont = cont;
BOOST_ASIO_CORO_REENTER(*this)
{
// Maybe suspend
- if(! d.ws.wr_block_)
+ if(d.ws.wr_block_.try_lock(this))
{
- // Acquire the write block
- d.ws.wr_block_ = d.tok;
-
// Make sure the stream is open
if(! d.ws.check_open(ec))
goto upcall;
@@ -138,19 +144,17 @@ operator()(
else
{
// Suspend
- BOOST_ASSERT(d.ws.wr_block_ != d.tok);
BOOST_ASIO_CORO_YIELD
d.ws.paused_close_.emplace(std::move(*this));
// Acquire the write block
- BOOST_ASSERT(! d.ws.wr_block_);
- d.ws.wr_block_ = d.tok;
+ d.ws.wr_block_.lock(this);
// Resume
BOOST_ASIO_CORO_YIELD
boost::asio::post(
d.ws.get_executor(), std::move(*this));
- BOOST_ASSERT(d.ws.wr_block_ == d.tok);
+ BOOST_ASSERT(d.ws.wr_block_.is_locked(this));
// Make sure the stream is open
if(! d.ws.check_open(ec))
@@ -181,27 +185,20 @@ operator()(
}
// Maybe suspend
- if(! d.ws.rd_block_)
- {
- // Acquire the read block
- d.ws.rd_block_ = d.tok;
- }
- else
+ if(! d.ws.rd_block_.try_lock(this))
{
// Suspend
- BOOST_ASSERT(d.ws.rd_block_ != d.tok);
BOOST_ASIO_CORO_YIELD
d.ws.paused_r_close_.emplace(std::move(*this));
// Acquire the read block
- BOOST_ASSERT(! d.ws.rd_block_);
- d.ws.rd_block_ = d.tok;
+ d.ws.rd_block_.lock(this);
// Resume
BOOST_ASIO_CORO_YIELD
boost::asio::post(
d.ws.get_executor(), std::move(*this));
- BOOST_ASSERT(d.ws.rd_block_ == d.tok);
+ BOOST_ASSERT(d.ws.rd_block_.is_locked(this));
// Make sure the stream is open
BOOST_ASSERT(d.ws.status_ != status::open);
@@ -219,13 +216,10 @@ operator()(
{
// Read frame header
while(! d.ws.parse_fh(
- d.ws.rd_fh_, d.ws.rd_buf_, code))
+ d.ws.rd_fh_, d.ws.rd_buf_, d.ev))
{
- if(code != close_code::none)
- {
- d.ev = error::failed;
+ if(d.ev)
goto teardown;
- }
BOOST_ASIO_CORO_YIELD
d.ws.stream_.async_read_some(
d.ws.rd_buf_.prepare(read_size(d.ws.rd_buf_,
@@ -244,16 +238,12 @@ operator()(
d.ws.rd_close_ = true;
auto const mb = buffers_prefix(
clamp(d.ws.rd_fh_.len),
- d.ws.rd_buf_.data());
+ d.ws.rd_buf_.mutable_data());
if(d.ws.rd_fh_.len > 0 && d.ws.rd_fh_.mask)
detail::mask_inplace(mb, d.ws.rd_key_);
- detail::read_close(d.ws.cr_, mb, code);
- if(code != close_code::none)
- {
- // Protocol error
- d.ev = error::failed;
+ detail::read_close(d.ws.cr_, mb, d.ev);
+ if(d.ev)
goto teardown;
- }
d.ws.rd_buf_.consume(clamp(d.ws.rd_fh_.len));
goto teardown;
}
@@ -283,12 +273,12 @@ operator()(
teardown:
// Teardown
- BOOST_ASSERT(d.ws.wr_block_ == d.tok);
+ BOOST_ASSERT(d.ws.wr_block_.is_locked(this));
using beast::websocket::async_teardown;
BOOST_ASIO_CORO_YIELD
async_teardown(d.ws.role_,
d.ws.stream_, std::move(*this));
- BOOST_ASSERT(d.ws.wr_block_ == d.tok);
+ BOOST_ASSERT(d.ws.wr_block_.is_locked(this));
if(ec == boost::asio::error::eof)
{
// Rationale:
@@ -304,13 +294,10 @@ operator()(
d.ws.close();
upcall:
- BOOST_ASSERT(d.ws.wr_block_ == d.tok);
- d.ws.wr_block_.reset();
- if(d.ws.rd_block_ == d.tok)
- {
- d.ws.rd_block_.reset();
+ BOOST_ASSERT(d.ws.wr_block_.is_locked(this));
+ d.ws.wr_block_.unlock(this);
+ if(d.ws.rd_block_.try_unlock(this))
d.ws.paused_r_rd_.maybe_invoke();
- }
d.ws.paused_rd_.maybe_invoke() ||
d.ws.paused_ping_.maybe_invoke() ||
d.ws.paused_wr_.maybe_invoke();
@@ -327,9 +314,9 @@ operator()(
//------------------------------------------------------------------------------
-template<class NextLayer>
+template<class NextLayer, bool deflateSupported>
void
-stream<NextLayer>::
+stream<NextLayer, deflateSupported>::
close(close_reason const& cr)
{
static_assert(is_sync_stream<next_layer_type>::value,
@@ -340,9 +327,9 @@ close(close_reason const& cr)
BOOST_THROW_EXCEPTION(system_error{ec});
}
-template<class NextLayer>
+template<class NextLayer, bool deflateSupported>
void
-stream<NextLayer>::
+stream<NextLayer, deflateSupported>::
close(close_reason const& cr, error_code& ec)
{
static_assert(is_sync_stream<next_layer_type>::value,
@@ -364,18 +351,18 @@ close(close_reason const& cr, error_code& ec)
if(! check_ok(ec))
return;
status_ = status::closing;
+ error_code result;
// Drain the connection
- close_code code{};
if(rd_remain_ > 0)
goto read_payload;
for(;;)
{
// Read frame header
- while(! parse_fh(rd_fh_, rd_buf_, code))
+ while(! parse_fh(rd_fh_, rd_buf_, result))
{
- if(code != close_code::none)
- return do_fail(close_code::none,
- error::failed, ec);
+ if(result)
+ return do_fail(
+ close_code::none, result, ec);
auto const bytes_transferred =
stream_.read_some(
rd_buf_.prepare(read_size(rd_buf_,
@@ -393,15 +380,15 @@ close(close_reason const& cr, error_code& ec)
rd_close_ = true;
auto const mb = buffers_prefix(
clamp(rd_fh_.len),
- rd_buf_.data());
+ rd_buf_.mutable_data());
if(rd_fh_.len > 0 && rd_fh_.mask)
detail::mask_inplace(mb, rd_key_);
- detail::read_close(cr_, mb, code);
- if(code != close_code::none)
+ detail::read_close(cr_, mb, result);
+ if(result)
{
- // Protocol error
- return do_fail(close_code::none,
- error::failed, ec);
+ // Protocol violation
+ return do_fail(
+ close_code::none, result, ec);
}
rd_buf_.consume(clamp(rd_fh_.len));
break;
@@ -434,20 +421,20 @@ close(close_reason const& cr, error_code& ec)
ec.assign(0, ec.category());
}
-template<class NextLayer>
+template<class NextLayer, bool deflateSupported>
template<class CloseHandler>
BOOST_ASIO_INITFN_RESULT_TYPE(
CloseHandler, void(error_code))
-stream<NextLayer>::
+stream<NextLayer, deflateSupported>::
async_close(close_reason const& cr, CloseHandler&& handler)
{
static_assert(is_async_stream<next_layer_type>::value,
"AsyncStream requirements not met");
- boost::asio::async_completion<CloseHandler,
- void(error_code)> init{handler};
+ BOOST_BEAST_HANDLER_INIT(
+ CloseHandler, void(error_code));
close_op<BOOST_ASIO_HANDLER_TYPE(
CloseHandler, void(error_code))>{
- init.completion_handler, *this, cr}(
+ std::move(init.completion_handler), *this, cr}(
{}, 0, false);
return init.result.get();
}