feedc0de/boost_beast
1
0
Fork 0
forked from boostorg/beast
Code Pull Requests Activity

Refactor stream operations and tests plus coverage

Browse Source
This commit is contained in:
Vinnie Falco
2017-08-31 17:52:36 -07:00
parent 9f089c2a33
commit 51a5a36118
22 changed files with 1290 additions and 354 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/boost/beast/websocket/detail/pausation.hpp
+6
View File
@@ -187,6 +187,12 @@ public:
void
save(F&& f);
explicit
operator bool() const
{
return base_ != nullptr;
}
bool
maybe_invoke()
{
include/boost/beast/websocket/detail/pmd_extension.hpp
+5 -39
View File
@@ -57,14 +57,17 @@ parse_bits(string_view s)
return -1;
if(s[0] < '1' || s[0] > '9')
return -1;
int i = 0;
unsigned i = 0;
for(auto c : s)
{
if(c < '0' || c > '9')
return -1;
auto const i0 = i;
i = 10 * i + (c - '0');
if(i < i0)
return -1;
}
return i;
return static_cast<int>(i);
}
// Parse permessage-deflate request fields
@@ -354,43 +357,6 @@ pmd_normalize(pmd_offer& offer)
//--------------------------------------------------------------------
// Decompress into a DynamicBuffer
//
template<class InflateStream, class DynamicBuffer>
void
inflate(
InflateStream& zi,
DynamicBuffer& buffer,
boost::asio::const_buffer const& in,
error_code& ec)
{
using boost::asio::buffer_cast;
using boost::asio::buffer_size;
zlib::z_params zs;
zs.avail_in = buffer_size(in);
zs.next_in = buffer_cast<void const*>(in);
for(;;)
{
// VFALCO we could be smarter about the size
auto const bs = buffer.prepare(
read_size_or_throw(buffer, 65536));
auto const out = *bs.begin();
zs.avail_out = buffer_size(out);
zs.next_out = buffer_cast<void*>(out);
zi.write(zs, zlib::Flush::sync, ec);
buffer.commit(zs.total_out);
zs.total_out = 0;
if( ec == zlib::error::need_buffers ||
ec == zlib::error::end_of_stream)
{
ec.assign(0, ec.category());
break;
}
if(ec)
return;
}
}
// Compress a buffer sequence
// Returns: `true` if more calls are needed
//
include/boost/beast/websocket/impl/close.ipp
+28 -20
View File
@@ -22,8 +22,6 @@
#include <boost/throw_exception.hpp>
#include <memory>
#include <iostream>
namespace boost {
namespace beast {
namespace websocket {
@@ -46,7 +44,7 @@ class stream<NextLayer>::close_op
detail::frame_buffer fb;
error_code ev;
token tok;
bool cont;
bool cont = false;
state(
Handler&,
@@ -121,7 +119,8 @@ public:
template<class NextLayer>
template<class Handler>
void
stream<NextLayer>::close_op<Handler>::
stream<NextLayer>::
close_op<Handler>::
operator()(
error_code ec,
std::size_t bytes_transferred,
@@ -140,7 +139,7 @@ operator()(
d.ws.wr_block_ = d.tok;
// Make sure the stream is open
if(d.ws.check_fail(ec))
if(! d.ws.check_open(ec))
goto upcall;
}
else
@@ -160,29 +159,33 @@ operator()(
BOOST_ASSERT(d.ws.wr_block_ == d.tok);
// Make sure the stream is open
if(d.ws.check_fail(ec))
if(! d.ws.check_open(ec))
goto upcall;
}
// Can't call close twice
BOOST_ASSERT(! d.ws.wr_close_);
// Change status to closing
BOOST_ASSERT(d.ws.status_ == status::open);
d.ws.status_ = status::closing;
// Send close frame
d.ws.wr_close_ = true;
BOOST_ASIO_CORO_YIELD
boost::asio::async_write(d.ws.stream_,
d.fb.data(), std::move(*this));
if(d.ws.check_fail(ec))
if(! d.ws.check_ok(ec))
goto upcall;
if(d.ws.rd_close_)
{
// This happens when the read_op gets a close frame
// at the same time we are sending the close frame. The
// read_op will be suspended on the write block.
// at the same time close_op is sending the close frame.
// The read_op will be suspended on the write block.
goto teardown;
}
// Maybe suspend
if(! d.ws.rd_block_)
{
@@ -206,7 +209,9 @@ operator()(
BOOST_ASSERT(d.ws.rd_block_ == d.tok);
// Make sure the stream is open
if(d.ws.check_fail(ec))
BOOST_ASSERT(d.ws.status_ != status::open);
BOOST_ASSERT(d.ws.status_ != status::closed);
if( d.ws.status_ == status::failed)
goto upcall;
BOOST_ASSERT(! d.ws.rd_close_);
@@ -231,7 +236,7 @@ operator()(
d.ws.rd_buf_.prepare(read_size(d.ws.rd_buf_,
d.ws.rd_buf_.max_size())),
std::move(*this));
if(d.ws.check_fail(ec))
if(! d.ws.check_ok(ec))
goto upcall;
d.ws.rd_buf_.commit(bytes_transferred);
}
@@ -271,7 +276,7 @@ operator()(
d.ws.rd_buf_.prepare(read_size(d.ws.rd_buf_,
d.ws.rd_buf_.max_size())),
std::move(*this));
if(d.ws.check_fail(ec))
if(! d.ws.check_ok(ec))
goto upcall;
d.ws.rd_buf_.commit(bytes_transferred);
}
@@ -289,7 +294,6 @@ operator()(
async_teardown(d.ws.role_,
d.ws.stream_, std::move(*this));
BOOST_ASSERT(d.ws.wr_block_ == d.tok);
BOOST_ASSERT(d.ws.open_);
if(ec == boost::asio::error::eof)
{
// Rationale:
@@ -298,7 +302,11 @@ operator()(
}
if(! ec)
ec = d.ev;
d.ws.open_ = false;
if(ec)
d.ws.status_ = status::failed;
else
d.ws.status_ = status::closed;
d.ws.close();
upcall:
BOOST_ASSERT(d.ws.wr_block_ == d.tok);
@@ -346,7 +354,7 @@ close(close_reason const& cr, error_code& ec)
using beast::detail::clamp;
ec.assign(0, ec.category());
// Make sure the stream is open
if(check_fail(ec))
if(! check_open(ec))
return;
// If rd_close_ is set then we already sent a close
BOOST_ASSERT(! rd_close_);
@@ -357,8 +365,9 @@ close(close_reason const& cr, error_code& ec)
write_close<flat_static_buffer_base>(fb, cr);
boost::asio::write(stream_, fb.data(), ec);
}
if(check_fail(ec))
if(! check_ok(ec))
return;
status_ = status::closing;
// Drain the connection
close_code code{};
if(rd_remain_ > 0)
@@ -375,7 +384,7 @@ close(close_reason const& cr, error_code& ec)
stream_.read_some(
rd_buf_.prepare(read_size(rd_buf_,
rd_buf_.max_size())), ec);
if(check_fail(ec))
if(! check_ok(ec))
return;
rd_buf_.commit(bytes_transferred);
}
@@ -414,8 +423,7 @@ close(close_reason const& cr, error_code& ec)
stream_.read_some(
rd_buf_.prepare(read_size(rd_buf_,
rd_buf_.max_size())), ec);
open_ = ! ec;
if(! open_)
if(! check_ok(ec))
return;
rd_buf_.commit(bytes_transferred);
}
include/boost/beast/websocket/impl/ping.ipp
+7 -9
View File
@@ -131,7 +131,7 @@ operator()(error_code ec, std::size_t)
d.ws.wr_block_ = d.tok;
// Make sure the stream is open
if(d.ws.check_fail(ec))
if(! d.ws.check_open(ec))
{
BOOST_ASIO_CORO_YIELD
d.ws.get_io_service().post(
@@ -156,7 +156,7 @@ operator()(error_code ec, std::size_t)
BOOST_ASSERT(d.ws.wr_block_ == d.tok);
// Make sure the stream is open
if(d.ws.check_fail(ec))
if(! d.ws.check_open(ec))
goto upcall;
}
@@ -164,7 +164,7 @@ operator()(error_code ec, std::size_t)
BOOST_ASIO_CORO_YIELD
boost::asio::async_write(d.ws.stream_,
d.fb.data(), std::move(*this));
if(d.ws.check_fail(ec))
if(! d.ws.check_ok(ec))
goto upcall;
upcall:
@@ -195,15 +195,14 @@ void
stream<NextLayer>::
ping(ping_data const& payload, error_code& ec)
{
ec.assign(0, ec.category());
// Make sure the stream is open
if(check_fail(ec))
if(! check_open(ec))
return;
detail::frame_buffer fb;
write_ping<flat_static_buffer_base>(
fb, detail::opcode::ping, payload);
boost::asio::write(stream_, fb.data(), ec);
if(check_fail(ec))
if(! check_ok(ec))
return;
}
@@ -223,15 +222,14 @@ void
stream<NextLayer>::
pong(ping_data const& payload, error_code& ec)
{
ec.assign(0, ec.category());
// Make sure the stream is open
if(check_fail(ec))
if(! check_open(ec))
return;
detail::frame_buffer fb;
write_ping<flat_static_buffer_base>(
fb, detail::opcode::pong, payload);
boost::asio::write(stream_, fb.data(), ec);
if(check_fail(ec))
if(! check_ok(ec))
return;
}
include/boost/beast/websocket/impl/read.ipp
+72 -54
View File
@@ -53,7 +53,7 @@ class stream<NextLayer>::read_some_op
token tok_;
close_code code_;
bool did_read_ = false;
bool cont_;
bool cont_ = false;
public:
read_some_op(read_some_op&&) = default;
@@ -138,14 +138,19 @@ operator()(
BOOST_ASIO_CORO_REENTER(*this)
{
// Maybe suspend
do_maybe_suspend:
if(! ws_.rd_block_)
{
// Acquire the read block
ws_.rd_block_ = tok_;
// Make sure the stream is open
if(ws_.check_fail(ec))
// Make sure the stream is not closed
if( ws_.status_ == status::closed ||
ws_.status_ == status::failed)
{
ec = boost::asio::error::operation_aborted;
goto upcall;
}
}
else
{
@@ -164,10 +169,18 @@ operator()(
ws_.get_io_service().post(std::move(*this));
BOOST_ASSERT(ws_.rd_block_ == tok_);
// Make sure the stream is open
if(ws_.check_fail(ec))
goto upcall;
// The only way to get read blocked is if
// a `close_op` wrote a close frame
BOOST_ASSERT(ws_.wr_close_);
BOOST_ASSERT(ws_.status_ != status::open);
ec = boost::asio::error::operation_aborted;
goto upcall;
}
// if status_ == status::closing, we want to suspend
// the read operation until the close completes,
// then finish the read with operation_aborted.
loop:
BOOST_ASSERT(ws_.rd_block_ == tok_);
// See if we need to read a frame header. This
@@ -195,19 +208,21 @@ operator()(
ws_.rd_buf_, ws_.rd_buf_.max_size())),
std::move(*this));
BOOST_ASSERT(ws_.rd_block_ == tok_);
if(ws_.check_fail(ec))
if(! ws_.check_ok(ec))
goto upcall;
ws_.rd_buf_.commit(bytes_transferred);
// Allow a close operation to
// drain the connection if necessary.
// Allow a close operation
// to acquire the read block
BOOST_ASSERT(ws_.rd_block_ == tok_);
ws_.rd_block_.reset();
if( ws_.paused_r_close_.maybe_invoke())
{
// Suspend
BOOST_ASSERT(ws_.rd_block_);
goto do_suspend;
}
// Acquire read block
ws_.rd_block_ = tok_;
}
// Immediately apply the mask to the portion
@@ -236,7 +251,7 @@ operator()(
detail::read_ping(payload, b);
ws_.rd_buf_.consume(len);
// Ignore ping when closing
if(ws_.wr_close_)
if(ws_.status_ == status::closing)
goto loop;
if(ws_.ctrl_cb_)
ws_.ctrl_cb_(frame_type::ping, payload);
@@ -245,6 +260,15 @@ operator()(
flat_static_buffer_base>(ws_.rd_fb_,
detail::opcode::pong, payload);
}
//BOOST_ASSERT(! ws_.paused_r_close_);
// Allow a close operation
// to acquire the read block
BOOST_ASSERT(ws_.rd_block_ == tok_);
ws_.rd_block_.reset();
ws_.paused_r_close_.maybe_invoke();
// Maybe suspend
if(! ws_.wr_block_)
{
@@ -268,18 +292,8 @@ operator()(
BOOST_ASSERT(ws_.wr_block_ == tok_);
// Make sure the stream is open
if(ws_.check_fail(ec))
if(! ws_.check_open(ec))
goto upcall;
// Ignore ping when closing
if(ws_.wr_close_)
{
ws_.wr_block_.reset();
ws_.paused_close_.maybe_invoke() ||
ws_.paused_ping_.maybe_invoke() ||
ws_.paused_wr_.maybe_invoke();
goto loop;
}
}
// Send pong
@@ -288,13 +302,13 @@ operator()(
boost::asio::async_write(ws_.stream_,
ws_.rd_fb_.data(), std::move(*this));
BOOST_ASSERT(ws_.wr_block_ == tok_);
if(ws_.check_fail(ec))
if(! ws_.check_ok(ec))
goto upcall;
ws_.wr_block_.reset();
ws_.paused_close_.maybe_invoke() ||
ws_.paused_ping_.maybe_invoke() ||
ws_.paused_wr_.maybe_invoke();
goto loop;
goto do_maybe_suspend;
}
// Handle pong frame
if(ws_.rd_fh_.op == detail::opcode::pong)
@@ -335,17 +349,19 @@ operator()(
if(ws_.ctrl_cb_)
ws_.ctrl_cb_(frame_type::close,
ws_.cr_.reason);
if(! ws_.wr_close_)
// See if we are already closing
if(ws_.status_ == status::closing)
{
// _Start the WebSocket Closing Handshake_
code_ = cr.code == close_code::none ?
close_code::normal :
static_cast<close_code>(cr.code);
// _Close the WebSocket Connection_
BOOST_ASSERT(ws_.wr_close_);
code_ = close_code::none;
ev_ = error::closed;
goto close;
}
// _Close the WebSocket Connection_
code_ = close_code::none;
// _Start the WebSocket Closing Handshake_
code_ = cr.code == close_code::none ?
close_code::normal :
static_cast<close_code>(cr.code);
ev_ = error::closed;
goto close;
}
@@ -372,7 +388,7 @@ operator()(
ws_.rd_buf_.prepare(read_size(
ws_.rd_buf_, ws_.rd_buf_.max_size())),
std::move(*this));
if(ws_.check_fail(ec))
if(! ws_.check_ok(ec))
goto upcall;
ws_.rd_buf_.commit(bytes_transferred);
if(ws_.rd_fh_.mask)
@@ -413,7 +429,7 @@ operator()(
BOOST_ASIO_CORO_YIELD
ws_.stream_.async_read_some(buffer_prefix(
clamp(ws_.rd_remain_), cb_), std::move(*this));
if(ws_.check_fail(ec))
if(! ws_.check_ok(ec))
goto upcall;
BOOST_ASSERT(bytes_transferred > 0);
auto const mb = buffer_prefix(
@@ -456,7 +472,7 @@ operator()(
ws_.rd_buf_.prepare(read_size(
ws_.rd_buf_, ws_.rd_buf_.max_size())),
std::move(*this));
if(ws_.check_fail(ec))
if(! ws_.check_ok(ec))
goto upcall;
BOOST_ASSERT(bytes_transferred > 0);
ws_.rd_buf_.commit(bytes_transferred);
@@ -498,8 +514,7 @@ operator()(
zs.next_in = empty_block;
zs.avail_in = sizeof(empty_block);
ws_.pmd_->zi.write(zs, zlib::Flush::sync, ec);
if(ws_.check_fail(ec))
goto upcall;
BOOST_ASSERT(! ec);
// VFALCO See:
// https://github.com/madler/zlib/issues/280
BOOST_ASSERT(zs.total_out == 0);
@@ -521,7 +536,7 @@ operator()(
}
ws_.pmd_->zi.write(zs, zlib::Flush::sync, ec);
BOOST_ASSERT(ec != zlib::error::end_of_stream);
if(ws_.check_fail(ec))
if(! ws_.check_ok(ec))
goto upcall;
if(ws_.rd_msg_max_ && beast::detail::sum_exceeds(
ws_.rd_size_, zs.total_out, ws_.rd_msg_max_))
@@ -560,7 +575,7 @@ operator()(
ws_.wr_block_ = tok_;
// Make sure the stream is open
BOOST_ASSERT(ws_.open_);
BOOST_ASSERT(ws_.status_ == status::open);
}
else
{
@@ -579,10 +594,13 @@ operator()(
BOOST_ASSERT(ws_.wr_block_ == tok_);
// Make sure the stream is open
if(ws_.check_fail(ec))
if(! ws_.check_open(ec))
goto upcall;
}
// Set the status
ws_.status_ = status::closing;
if(! ws_.wr_close_)
{
ws_.wr_close_ = true;
@@ -600,9 +618,7 @@ operator()(
ws_.stream_, ws_.rd_fb_.data(),
std::move(*this));
BOOST_ASSERT(ws_.wr_block_ == tok_);
// Make sure the stream is open
if(ws_.check_fail(ec))
if(! ws_.check_ok(ec))
goto upcall;
}
@@ -621,11 +637,15 @@ operator()(
}
if(! ec)
ec = ev_;
ws_.open_ = false;
if(ec && ec != error::closed)
ws_.status_ = status::failed;
else
ws_.status_ = status::closed;
ws_.close();
upcall:
BOOST_ASSERT(ws_.rd_block_ == tok_);
ws_.rd_block_.reset();
if(ws_.rd_block_ == tok_)
ws_.rd_block_.reset();
ws_.paused_r_close_.maybe_invoke();
if(ws_.wr_block_ == tok_)
{
@@ -755,7 +775,7 @@ operator()(
BOOST_ASIO_CORO_YIELD
read_some_op<buffers_type, read_op>{
std::move(*this), ws_, *mb}(
{}, 0, true);
{}, 0, false);
if(ec)
break;
b_.commit(bytes_transferred);
@@ -956,7 +976,7 @@ read_some(
std::size_t bytes_written = 0;
ec.assign(0, ec.category());
// Make sure the stream is open
if(check_fail(ec))
if(! check_open(ec))
return 0;
loop:
// See if we need to read a frame header. This
@@ -979,7 +999,7 @@ loop:
rd_buf_.prepare(read_size(
rd_buf_, rd_buf_.max_size())),
ec);
if(check_fail(ec))
if(! check_ok(ec))
return bytes_written;
rd_buf_.commit(bytes_transferred);
}
@@ -1018,7 +1038,7 @@ loop:
write_ping<flat_static_buffer_base>(fb,
detail::opcode::pong, payload);
boost::asio::write(stream_, fb.data(), ec);
if(check_fail(ec))
if(! check_ok(ec))
return bytes_written;
goto loop;
}
@@ -1082,7 +1102,7 @@ loop:
rd_buf_.commit(stream_.read_some(
rd_buf_.prepare(read_size(rd_buf_,
rd_buf_.max_size())), ec));
if(check_fail(ec))
if(! check_ok(ec))
return bytes_written;
if(rd_fh_.mask)
detail::mask_inplace(
@@ -1125,7 +1145,7 @@ loop:
auto const bytes_transferred =
stream_.read_some(buffer_prefix(
clamp(rd_remain_), buffers), ec);
if(check_fail(ec))
if(! check_ok(ec))
return bytes_written;
BOOST_ASSERT(bytes_transferred > 0);
auto const mb = buffer_prefix(
@@ -1187,7 +1207,7 @@ loop:
rd_buf_.prepare(read_size(
rd_buf_, rd_buf_.max_size())),
ec);
if(check_fail(ec))
if(! check_ok(ec))
return bytes_written;
BOOST_ASSERT(bytes_transferred > 0);
rd_buf_.commit(bytes_transferred);
@@ -1217,8 +1237,6 @@ loop:
zs.avail_in = sizeof(empty_block);
pmd_->zi.write(zs, zlib::Flush::sync, ec);
BOOST_ASSERT(! ec);
if(check_fail(ec))
return bytes_written;
// VFALCO See:
// https://github.com/madler/zlib/issues/280
BOOST_ASSERT(zs.total_out == 0);
@@ -1240,7 +1258,7 @@ loop:
}
pmd_->zi.write(zs, zlib::Flush::sync, ec);
BOOST_ASSERT(ec != zlib::error::end_of_stream);
if(check_fail(ec))
if(! check_ok(ec))
return bytes_written;
if(rd_msg_max_ && beast::detail::sum_exceeds(
rd_size_, zs.total_out, rd_msg_max_))
include/boost/beast/websocket/impl/rfc6455.ipp
-2
View File
@@ -30,8 +30,6 @@ is_upgrade(http::header<true,
return false;
if(! http::token_list{req["Upgrade"]}.exists("websocket"))
return false;
if(! req.count(http::field::sec_websocket_version))
return false;
return true;
}
include/boost/beast/websocket/impl/stream.ipp
+11 -12
View File
@@ -134,7 +134,7 @@ open(role_type role)
{
// VFALCO TODO analyze and remove dupe code in reset()
role_ = role;
open_ = true;
status_ = status::open;
rd_remain_ = 0;
rd_cont_ = false;
rd_done_ = true;
@@ -193,8 +193,7 @@ void
stream<NextLayer>::
reset()
{
BOOST_ASSERT(! open_);
open_ = false; // VFALCO is this needed?
BOOST_ASSERT(status_ != status::open);
rd_remain_ = 0;
rd_cont_ = false;
rd_done_ = true;
@@ -593,8 +592,6 @@ build_response(http::request<Body,
return err("Missing Host");
if(! req.count(http::field::sec_websocket_key))
return err("Missing Sec-WebSocket-Key");
if(! http::token_list{req[http::field::upgrade]}.exists("websocket"))
return err("Missing websocket Upgrade token");
auto const key = req[http::field::sec_websocket_key];
if(key.size() > detail::sec_ws_key_type::max_size_n)
return err("Invalid Sec-WebSocket-Key");
@@ -684,6 +681,7 @@ do_fail(
error_code& ec) // set to the error, else set to ev
{
BOOST_ASSERT(ev);
status_ = status::closing;
if(code != close_code::none && ! wr_close_)
{
wr_close_ = true;
@@ -691,8 +689,7 @@ do_fail(
write_close<
flat_static_buffer_base>(fb, code);
boost::asio::write(stream_, fb.data(), ec);
open_ = ! ec;
if(! open_)
if(! check_ok(ec))
return;
}
using beast::websocket::teardown;
@@ -703,11 +700,13 @@ do_fail(
// http://stackoverflow.com/questions/25587403/boost-asio-ssl-async-shutdown-always-finishes-with-an-error
ec.assign(0, ec.category());
}
open_ = ! ec;
if(! open_)
return;
ec = ev;
open_ = false;
if(! ec)
ec = ev;
if(ec && ec != error::closed)
status_ = status::failed;
else
status_ = status::closed;
close();
}
} // websocket
include/boost/beast/websocket/impl/write.ipp
+18 -32
View File
@@ -49,7 +49,7 @@ class stream<NextLayer>::write_some_op
int how_;
bool fin_;
bool more_;
bool cont_;
bool cont_ = false;
public:
write_some_op(write_some_op&&) = default;
@@ -206,7 +206,7 @@ operator()(
ws_.wr_block_ = tok_;
// Make sure the stream is open
if(ws_.check_fail(ec))
if(! ws_.check_open(ec))
goto upcall;
}
else
@@ -227,7 +227,7 @@ operator()(
BOOST_ASSERT(ws_.wr_block_ == tok_);
// Make sure the stream is open
if(ws_.check_fail(ec))
if(! ws_.check_open(ec))
goto upcall;
}
@@ -242,13 +242,11 @@ operator()(
ws_.wr_fb_, fh_);
ws_.wr_cont_ = ! fin_;
// Send frame
BOOST_ASSERT(ws_.wr_block_ == tok_);
BOOST_ASIO_CORO_YIELD
boost::asio::async_write(ws_.stream_,
buffer_cat(ws_.wr_fb_.data(), cb_),
std::move(*this));
BOOST_ASSERT(ws_.wr_block_ == tok_);
if(ws_.check_fail(ec))
if(! ws_.check_ok(ec))
goto upcall;
goto upcall;
}
@@ -267,15 +265,13 @@ operator()(
ws_.wr_fb_, fh_);
ws_.wr_cont_ = ! fin_;
// Send frame
BOOST_ASSERT(ws_.wr_block_ == tok_);
BOOST_ASIO_CORO_YIELD
boost::asio::async_write(
ws_.stream_, buffer_cat(
ws_.wr_fb_.data(), buffer_prefix(
clamp(fh_.len), cb_)),
std::move(*this));
BOOST_ASSERT(ws_.wr_block_ == tok_);
if(ws_.check_fail(ec))
if(! ws_.check_ok(ec))
goto upcall;
if(remain_ == 0)
break;
@@ -317,14 +313,12 @@ operator()(
remain_ -= n;
ws_.wr_cont_ = ! fin_;
// Send frame header and partial payload
BOOST_ASSERT(ws_.wr_block_ == tok_);
BOOST_ASIO_CORO_YIELD
boost::asio::async_write(
ws_.stream_, buffer_cat(ws_.wr_fb_.data(),
buffer(ws_.wr_buf_.get(), n)),
std::move(*this));
BOOST_ASSERT(ws_.wr_block_ == tok_);
if(ws_.check_fail(ec))
if(! ws_.check_ok(ec))
goto upcall;
while(remain_ > 0)
{
@@ -336,13 +330,11 @@ operator()(
ws_.wr_buf_.get(), n), key_);
remain_ -= n;
// Send partial payload
BOOST_ASSERT(ws_.wr_block_ == tok_);
BOOST_ASIO_CORO_YIELD
boost::asio::async_write(ws_.stream_,
buffer(ws_.wr_buf_.get(), n),
std::move(*this));
BOOST_ASSERT(ws_.wr_block_ == tok_);
if(ws_.check_fail(ec))
if(! ws_.check_ok(ec))
goto upcall;
}
goto upcall;
@@ -369,14 +361,12 @@ operator()(
ws_.wr_fb_, fh_);
ws_.wr_cont_ = ! fin_;
// Send frame
BOOST_ASSERT(ws_.wr_block_ == tok_);
BOOST_ASIO_CORO_YIELD
boost::asio::async_write(ws_.stream_,
buffer_cat(ws_.wr_fb_.data(),
buffer(ws_.wr_buf_.get(), n)),
std::move(*this));
BOOST_ASSERT(ws_.wr_block_ == tok_);
if(ws_.check_fail(ec))
if(! ws_.check_ok(ec))
goto upcall;
if(remain_ == 0)
break;
@@ -408,7 +398,7 @@ operator()(
ws_.wr_buf_size_);
more_ = detail::deflate(
ws_.pmd_->zo, b, cb_, fin_, ec);
if(ws_.check_fail(ec))
if(! ws_.check_ok(ec))
goto upcall;
n = buffer_size(b);
if(n == 0)
@@ -434,13 +424,11 @@ operator()(
flat_static_buffer_base>(ws_.wr_fb_, fh_);
ws_.wr_cont_ = ! fin_;
// Send frame
BOOST_ASSERT(ws_.wr_block_ == tok_);
BOOST_ASIO_CORO_YIELD
boost::asio::async_write(ws_.stream_,
buffer_cat(ws_.wr_fb_.data(),
mutable_buffers_1{b}), std::move(*this));
BOOST_ASSERT(ws_.wr_block_ == tok_);
if(ws_.check_fail(ec))
if(! ws_.check_ok(ec))
goto upcall;
if(more_)
{
@@ -460,7 +448,6 @@ operator()(
}
else
{
BOOST_ASSERT(ws_.wr_block_ == tok_);
if(fh_.fin && (
(ws_.role_ == role_type::client &&
ws_.pmd_config_.client_no_context_takeover) ||
@@ -524,7 +511,7 @@ write_some(bool fin,
using boost::asio::buffer_size;
ec.assign(0, ec.category());
// Make sure the stream is open
if(check_fail(ec))
if(! check_open(ec))
return;
detail::frame_header fh;
if(! wr_cont_)
@@ -552,8 +539,7 @@ write_some(bool fin,
wr_buf_.get(), wr_buf_size_);
auto const more = detail::deflate(
pmd_->zo, b, cb, fin, ec);
open_ = ! ec;
if(! open_)
if(! check_ok(ec))
return;
auto const n = buffer_size(b);
if(n == 0)
@@ -581,7 +567,7 @@ write_some(bool fin,
wr_cont_ = ! fin;
boost::asio::write(stream_,
buffer_cat(fh_buf.data(), b), ec);
if(check_fail(ec))
if(! check_ok(ec))
return;
if(! more)
break;
@@ -608,7 +594,7 @@ write_some(bool fin,
wr_cont_ = ! fin;
boost::asio::write(stream_,
buffer_cat(fh_buf.data(), buffers), ec);
if(check_fail(ec))
if(! check_ok(ec))
return;
}
else
@@ -630,7 +616,7 @@ write_some(bool fin,
boost::asio::write(stream_,
buffer_cat(fh_buf.data(),
buffer_prefix(n, cb)), ec);
if(check_fail(ec))
if(! check_ok(ec))
return;
if(remain == 0)
break;
@@ -662,7 +648,7 @@ write_some(bool fin,
wr_cont_ = ! fin;
boost::asio::write(stream_,
buffer_cat(fh_buf.data(), b), ec);
if(check_fail(ec))
if(! check_ok(ec))
return;
}
while(remain > 0)
@@ -674,7 +660,7 @@ write_some(bool fin,
remain -= n;
detail::mask_inplace(b, key);
boost::asio::write(stream_, b, ec);
if(check_fail(ec))
if(! check_ok(ec))
return;
}
}
@@ -702,7 +688,7 @@ write_some(bool fin,
flat_static_buffer_base>(fh_buf, fh);
boost::asio::write(stream_,
buffer_cat(fh_buf.data(), b), ec);
if(check_fail(ec))
if(! check_ok(ec))
return;
if(remain == 0)
break;
include/boost/beast/websocket/stream.hpp
+27 -11
View File
@@ -132,6 +132,9 @@ class stream
struct op {};
using control_cb_type =
std::function<void(frame_type, string_view)>;
// tokens are used to order reads and writes
class token
{
@@ -157,8 +160,13 @@ class stream
zlib::inflate_stream zi;
};
using control_cb_type =
std::function<void(frame_type, string_view)>;
enum class status
{
open,
closing,
closed,
failed
};
NextLayer stream_; // the wrapped stream
close_reason cr_; // set from received close frame
@@ -190,8 +198,8 @@ class stream
token tok_; // used to order asynchronous ops
role_type role_ // server or client
= role_type::client;
bool open_ // `true` if connected
= false;
status status_
= status::closed;
token wr_block_; // op currenly writing
bool wr_close_ // did we write a close frame?
@@ -366,7 +374,7 @@ public:
bool
is_open() const
{
return open_;
return status_ == status::open;
}
/** Returns `true` if the latest message data indicates binary.
@@ -3394,19 +3402,27 @@ private:
void begin_msg();
bool
check_fail(error_code& ec)
check_open(error_code& ec)
{
if(! open_)
if(status_ != status::open)
{
ec = boost::asio::error::operation_aborted;
return true;
return false;
}
ec.assign(0, ec.category());
return true;
}
bool
check_ok(error_code& ec)
{
if(ec)
{
open_ = false;
return true;
if(status_ != status::closed)
status_ = status::failed;
return false;
}
return false;
return true;
}
template<class DynamicBuffer>