ARCHIVE/xahaud
3
0
Fork 0
mirror of https://github.com/Xahau/xahaud.git synced 2025-11-27 22:15:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

WebSocket ping, fixes, coverage:

Browse Source 
* Improve test coverage
* tests for invokable in composed ops

* Update documentation
* Add License badge to README
* Target Windows 7 SDK and later
* Make role_type private
* Remove extra unused masking functions
* Allow stream reuse / reconnect after failure
* Restructure logic of composed operations
* Allow 0 for read_message_max meaning no limit
* Respect keep alive when building HTTP responses
* Check version in upgrade request
* Response with 426 status on unsupported WebSocket version
* Remove unnecessary Sec-WebSocket-Key in HTTP responses
* Rename to mask_buffer_size

* Remove maybe_throw
* Add ping, async_ping, async_on_pong
* Add ping_op
* Add pong_op
* Fix crash in accept_op
* Fix suspend in close_op
* Fix read_frame_op logic
* Fix crash in read_op
* Fix races in echo sync and async echo servers
This commit is contained in:
Vinnie Falco
2016-05-15 16:22:25 -04:00
parent a570b74038
commit eb7bd6a2f1
40 changed files with 2757 additions and 1365 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
include/beast/websocket/impl/accept_op.ipp
View File

@@ -48,6 +48,7 @@ class stream<NextLayer>::accept_op
{
using boost::asio::buffer_copy;
using boost::asio::buffer_size;
ws.reset();
ws.stream_.buffer().commit(buffer_copy(
ws.stream_.buffer().prepare(
buffer_size(buffers)), buffers));
@@ -133,8 +134,14 @@ operator()(error_code const& ec,
// got message
case 1:
// respond to request
#if 1
// VFALCO I have no idea why passing std::move(*this) crashes
d.state = 99;
d.ws.async_accept(d.req, *this);
#else
response_op<Handler>{
std::move(d.h), d.ws, d.req, true};
#endif
return;
}
}

94
include/beast/websocket/impl/close_op.ipp
View File

@@ -21,12 +21,9 @@ template<class NextLayer>
template<class Handler>
class stream<NextLayer>::close_op
{
using alloc_type =
handler_alloc<char, Handler>;
using fb_type =
detail::frame_streambuf;
using fmb_type =
typename fb_type::mutable_buffers_type;
using alloc_type = handler_alloc<char, Handler>;
using fb_type = detail::frame_streambuf;
struct data : op
{
@@ -64,16 +61,19 @@ public:
std::forward<DeducedHandler>(h), ws,
std::forward<Args>(args)...))
{
(*this)(error_code{}, 0, false);
(*this)(error_code{}, false);
}
void operator()()
{
(*this)(error_code{}, 0, true);
(*this)(error_code{});
}
void
operator()(error_code ec, std::size_t, bool again = true);
operator()(error_code ec, std::size_t);
void
operator()(error_code ec, bool again = true);
friend
void* asio_handler_allocate(
@@ -110,11 +110,25 @@ template<class NextLayer>
template<class Handler>
void
stream<NextLayer>::close_op<Handler>::
operator()(error_code ec, std::size_t, bool again)
operator()(error_code ec, std::size_t)
{
auto& d = *d_;
if(ec)
d.ws.failed_ = true;
(*this)(ec);
}
template<class NextLayer>
template<class Handler>
void
stream<NextLayer>::close_op<Handler>::
operator()(error_code ec, bool again)
{
auto& d = *d_;
d.cont = d.cont || again;
while(! ec && d.state != 99)
if(ec)
goto upcall;
for(;;)
{
switch(d.state)
{
@@ -122,58 +136,52 @@ operator()(error_code ec, std::size_t, bool again)
if(d.ws.wr_block_)
{
// suspend
d.state = 1;
d.ws.rd_op_.template emplace<
d.state = 2;
d.ws.wr_op_.template emplace<
close_op>(std::move(*this));
return;
}
if(d.ws.error_)
if(d.ws.failed_ || d.ws.wr_close_)
{
// call handler
d.state = 99;
d.ws.get_io_service().post(
bind_handler(std::move(*this),
boost::asio::error::operation_aborted, 0));
boost::asio::error::operation_aborted));
return;
}
d.state = 3;
break;
// fall through
// resume
case 1:
// VFALCO NOTE Should d.cont be `true` or false here?
// Does this count as a continuation of the original call
// to the asynchronous initiation function (async_close)?
d.state = 2;
d.ws.get_io_service().post(bind_handler(
std::move(*this), ec, 0));
return;
case 2:
if(d.ws.error_)
{
// call handler
d.state = 99;
ec = boost::asio::error::operation_aborted;
break;
}
d.state = 3; // VFALCO fall through?
break;
case 3:
// send close
// send close frame
d.state = 99;
assert(! d.ws.wr_close_);
d.ws.wr_close_ = true;
assert(! d.ws.wr_block_);
d.ws.wr_block_ = &d;
boost::asio::async_write(d.ws.stream_,
d.fb.data(), std::move(*this));
return;
case 2:
d.state = 3;
d.ws.get_io_service().post(
bind_handler(std::move(*this), ec));
return;
case 3:
if(d.ws.failed_ || d.ws.wr_close_)
{
// call handler
ec = boost::asio::error::operation_aborted;
goto upcall;
}
d.state = 1;
break;
case 99:
goto upcall;
}
}
if(ec)
d.ws.error_ = true;
upcall:
if(d.ws.wr_block_ == &d)
d.ws.wr_block_ = nullptr;
d.ws.rd_op_.maybe_invoke();

1
include/beast/websocket/impl/handshake_op.ipp
View File

@@ -48,6 +48,7 @@ class stream<NextLayer>::handshake_op
, cont(boost_asio_handler_cont_helpers::
is_continuation(h))
{
ws.reset();
}
};

193
include/beast/websocket/impl/ping_op.ipp Normal file
View File

@@ -0,0 +1,193 @@
//
// Copyright (c) 2013-2016 Vinnie Falco (vinnie dot falco at gmail dot com)
//
// Distributed under the Boost Software License, Version 1.0. (See accompanying
// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
//
#ifndef BEAST_WEBSOCKET_IMPL_PING_OP_HPP
#define BEAST_WEBSOCKET_IMPL_PING_OP_HPP
#include <beast/core/bind_handler.hpp>
#include <beast/core/handler_alloc.hpp>
#include <beast/websocket/detail/frame.hpp>
#include <memory>
namespace beast {
namespace websocket {
// write a ping frame
//
template<class NextLayer>
template<class Handler>
class stream<NextLayer>::ping_op
{
using alloc_type =
handler_alloc<char, Handler>;
struct data : op
{
stream<NextLayer>& ws;
Handler h;
detail::frame_streambuf fb;
bool cont;
int state = 0;
template<class DeducedHandler>
data(DeducedHandler&& h_, stream<NextLayer>& ws_,
ping_data const& payload)
: ws(ws_)
, h(std::forward<DeducedHandler>(h_))
, cont(boost_asio_handler_cont_helpers::
is_continuation(h))
{
using boost::asio::buffer;
using boost::asio::buffer_copy;
ws.template write_ping<static_streambuf>(
fb, opcode::ping, payload);
}
};
std::shared_ptr<data> d_;
public:
ping_op(ping_op&&) = default;
ping_op(ping_op const&) = default;
template<class DeducedHandler, class... Args>
ping_op(DeducedHandler&& h,
stream<NextLayer>& ws, Args&&... args)
: d_(std::make_shared<data>(
std::forward<DeducedHandler>(h), ws,
std::forward<Args>(args)...))
{
(*this)(error_code{}, false);
}
void operator()()
{
(*this)(error_code{});
}
void operator()(error_code ec, std::size_t);
void operator()(error_code ec, bool again = true);
friend
void* asio_handler_allocate(
std::size_t size, ping_op* op)
{
return boost_asio_handler_alloc_helpers::
allocate(size, op->d_->h);
}
friend
void asio_handler_deallocate(
void* p, std::size_t size, ping_op* op)
{
return boost_asio_handler_alloc_helpers::
deallocate(p, size, op->d_->h);
}
friend
bool asio_handler_is_continuation(ping_op* op)
{
return op->d_->cont;
}
template <class Function>
friend
void asio_handler_invoke(Function&& f, ping_op* op)
{
return boost_asio_handler_invoke_helpers::
invoke(f, op->d_->h);
}
};
template<class NextLayer>
template<class Handler>
void
stream<NextLayer>::ping_op<Handler>::
operator()(error_code ec, std::size_t)
{
auto& d = *d_;
if(ec)
d.ws.failed_ = true;
(*this)(ec);
}
template<class NextLayer>
template<class Handler>
void
stream<NextLayer>::
ping_op<Handler>::
operator()(error_code ec, bool again)
{
auto& d = *d_;
d.cont = d.cont || again;
if(ec)
goto upcall;
for(;;)
{
switch(d.state)
{
case 0:
if(d.ws.wr_block_)
{
// suspend
d.state = 2;
d.ws.wr_op_.template emplace<
ping_op>(std::move(*this));
return;
}
if(d.ws.failed_ || d.ws.wr_close_)
{
// call handler
d.state = 99;
d.ws.get_io_service().post(
bind_handler(std::move(*this),
boost::asio::error::operation_aborted));
return;
}
// fall through
case 1:
// send ping frame
d.state = 99;
assert(! d.ws.wr_block_);
d.ws.wr_block_ = &d;
boost::asio::async_write(d.ws.stream_,
d.fb.data(), std::move(*this));
return;
case 2:
d.state = 3;
d.ws.get_io_service().post(
bind_handler(std::move(*this), ec));
return;
case 3:
if(d.ws.failed_ || d.ws.wr_close_)
{
// call handler
ec = boost::asio::error::operation_aborted;
goto upcall;
}
d.state = 1;
break;
case 99:
goto upcall;
}
}
upcall:
if(d.ws.wr_block_ == &d)
d.ws.wr_block_ = nullptr;
d.ws.rd_op_.maybe_invoke();
d.h(ec);
}
} // websocket
} // beast
#endif

687
include/beast/websocket/impl/read_frame_op.ipp
View File

@@ -86,11 +86,14 @@ public:
void operator()(error_code const& ec)
{
(*this)(ec, 0);
(*this)(ec, 0, true);
}
void operator()(error_code ec,
std::size_t bytes_transferred, bool again = true);
std::size_t bytes_transferred);
void operator()(error_code ec,
std::size_t bytes_transferred, bool again);
friend
void* asio_handler_allocate(
@@ -127,393 +130,415 @@ template<class NextLayer>
template<class Buffers, class Handler>
void
stream<NextLayer>::read_frame_op<Buffers, Handler>::
operator()(error_code ec,std::size_t bytes_transferred, bool again)
operator()(error_code ec, std::size_t bytes_transferred)
{
auto& d = *d_;
d.cont = d.cont || again;
close_code::value code = close_code::none;
while(! ec && d.state != 99)
if(ec)
d.ws.failed_ = true;
(*this)(ec, bytes_transferred, true);
}
template<class NextLayer>
template<class Buffers, class Handler>
void
stream<NextLayer>::read_frame_op<Buffers, Handler>::
operator()(error_code ec,std::size_t bytes_transferred, bool again)
{
enum
{
switch(d.state)
do_start = 0,
do_read_payload = 1,
do_frame_done = 3,
do_read_fh = 4,
do_control_payload = 7,
do_control = 8,
do_pong_resume = 9,
do_pong = 11,
do_close_resume = 13,
do_close = 15,
do_fail = 18,
do_call_handler = 99
};
auto& d = *d_;
if(! ec)
{
d.cont = d.cont || again;
close_code::value code = close_code::none;
do
{
case 0:
if(d.ws.error_)
switch(d.state)
{
case do_start:
if(d.ws.failed_)
{
d.state = do_call_handler;
d.ws.get_io_service().post(
bind_handler(std::move(*this),
boost::asio::error::operation_aborted, 0));
return;
}
d.state = d.ws.rd_need_ > 0 ?
do_read_payload : do_read_fh;
break;
//------------------------------------------------------------------
case do_read_payload:
d.state = do_read_payload + 1;
d.smb = d.sb.prepare(
detail::clamp(d.ws.rd_need_));
// receive payload data
d.ws.stream_.async_read_some(
*d.smb, std::move(*this));
return;
case do_read_payload + 1:
{
d.ws.rd_need_ -= bytes_transferred;
auto const pb = prepare_buffers(
bytes_transferred, *d.smb);
if(d.ws.rd_fh_.mask)
detail::mask_inplace(pb, d.ws.rd_key_);
if(d.ws.rd_opcode_ == opcode::text)
{
if(! d.ws.rd_utf8_check_.write(pb) ||
(d.ws.rd_need_ == 0 && d.ws.rd_fh_.fin &&
! d.ws.rd_utf8_check_.finish()))
{
// invalid utf8
code = close_code::bad_payload;
d.state = do_fail;
break;
}
}
d.sb.commit(bytes_transferred);
if(d.ws.rd_need_ > 0)
{
d.state = do_read_payload;
break;
}
// fall through
}
//------------------------------------------------------------------
case do_frame_done:
// call handler
d.state = 99;
d.ws.get_io_service().post(
bind_handler(std::move(*this),
boost::asio::error::operation_aborted, 0));
d.fi.op = d.ws.rd_opcode_;
d.fi.fin = d.ws.rd_fh_.fin &&
d.ws.rd_need_ == 0;
goto upcall;
//------------------------------------------------------------------
case do_read_fh:
d.state = do_read_fh + 1;
boost::asio::async_read(d.ws.stream_,
d.fb.prepare(2), std::move(*this));
return;
case do_read_fh + 1:
{
d.fb.commit(bytes_transferred);
code = close_code::none;
auto const n = detail::read_fh1(
d.ws.rd_fh_, d.fb, d.ws.role_, code);
if(code != close_code::none)
{
// protocol error
d.state = do_fail;
break;
}
d.state = do_read_fh + 2;
if (n == 0)
{
bytes_transferred = 0;
break;
}
// read variable header
boost::asio::async_read(d.ws.stream_,
d.fb.prepare(n), std::move(*this));
return;
}
if(d.ws.rd_need_ > 0)
{
d.state = 1;
break;
}
d.state = 2;
break;
case 1:
// read payload
d.state = 3;
d.smb = d.sb.prepare(
detail::clamp(d.ws.rd_need_));
d.ws.stream_.async_read_some(
*d.smb, std::move(*this));
return;
case 2:
// read fixed header
d.state = 5;
boost::asio::async_read(d.ws.stream_,
d.fb.prepare(2), std::move(*this));
return;
// got payload
case 3:
{
d.ws.rd_need_ -= bytes_transferred;
auto const pb = prepare_buffers(
bytes_transferred, *d.smb);
if(d.ws.rd_fh_.mask)
detail::mask_inplace(pb, d.ws.rd_key_);
if(d.ws.rd_opcode_ == opcode::text)
{
if(! d.ws.rd_utf8_check_.write(pb) ||
(d.ws.rd_need_ == 0 && d.ws.rd_fh_.fin &&
! d.ws.rd_utf8_check_.finish()))
{
// invalid utf8
d.state = 18;
code = close_code::bad_payload;
break;
}
}
d.sb.commit(bytes_transferred);
d.state = 4;
break;
}
// call handler
case 4:
d.state = 99;
d.fi.op = d.ws.rd_opcode_;
d.fi.fin = d.ws.rd_fh_.fin &&
d.ws.rd_need_ == 0;
break;
// got fixed header
case 5:
{
d.fb.commit(bytes_transferred);
code = close_code::none;
auto const n = detail::read_fh1(
d.ws.rd_fh_, d.fb, d.ws.role_, code);
if(code != close_code::none)
{
// protocol error
d.state = 18;
break;
}
d.state = 6;
if (n == 0)
{
bytes_transferred = 0;
break;
}
// read variable header
boost::asio::async_read(d.ws.stream_,
d.fb.prepare(n), std::move(*this));
return;
}
// got variable header
case 6:
d.fb.commit(bytes_transferred);
code = close_code::none;
detail::read_fh2(d.ws.rd_fh_,
d.fb, d.ws.role_, code);
if(code == close_code::none)
d.ws.prepare_fh(code);
if(code != close_code::none)
{
// protocol error
d.state = 18;
break;
}
if(detail::is_control(d.ws.rd_fh_.op))
{
if(d.ws.rd_fh_.len > 0)
{
// read control payload
d.state = 7;
d.fmb = d.fb.prepare(static_cast<
std::size_t>(d.ws.rd_fh_.len));
boost::asio::async_read(d.ws.stream_,
*d.fmb, std::move(*this));
return;
}
d.state = 8;
break;
}
if(d.ws.rd_need_ > 0)
{
d.state = 1;
break;
}
if(! d.ws.rd_fh_.fin)
{
d.state = 2;
break;
}
// empty frame with fin
d.state = 4;
break;
// got control payload
case 7:
if(d.ws.rd_fh_.mask)
detail::mask_inplace(
*d.fmb, d.ws.rd_key_);
d.fb.commit(bytes_transferred);
d.state = 8;
break;
// do control
case 8:
if(d.ws.rd_fh_.op == opcode::ping)
{
case do_read_fh + 2:
d.fb.commit(bytes_transferred);
code = close_code::none;
ping_payload_type data;
detail::read(data, d.fb.data(), code);
detail::read_fh2(d.ws.rd_fh_,
d.fb, d.ws.role_, code);
if(code == close_code::none)
d.ws.prepare_fh(code);
if(code != close_code::none)
{
// protocol error
d.state = 18;
d.state = do_fail;
break;
}
d.fb.reset();
if(d.ws.wr_close_)
if(detail::is_control(d.ws.rd_fh_.op))
{
d.state = 2;
if(d.ws.rd_fh_.len > 0)
{
// read control payload
d.state = do_control_payload;
d.fmb = d.fb.prepare(static_cast<
std::size_t>(d.ws.rd_fh_.len));
boost::asio::async_read(d.ws.stream_,
*d.fmb, std::move(*this));
return;
}
d.state = do_control;
break;
}
d.ws.template write_ping<static_streambuf>(
d.fb, opcode::pong, data);
if(d.ws.wr_block_)
if(d.ws.rd_need_ > 0)
{
assert(d.ws.wr_block_ != &d);
// suspend
d.state = 13;
d.ws.rd_op_.template emplace<
read_frame_op>(std::move(*this));
return;
d.state = do_read_payload;
break;
}
d.state = 14;
// empty frame
d.state = do_frame_done;
break;
}
else if(d.ws.rd_fh_.op == opcode::pong)
{
code = close_code::none;
ping_payload_type data;
detail::read(data, d.fb.data(), code);
if(code != close_code::none)
{
// protocol error
d.state = 18;
break;
}
d.fb.reset();
// VFALCO TODO maybe_invoke an async pong handler
// For now just ignore the pong.
d.state = 2;
//------------------------------------------------------------------
case do_control_payload:
if(d.ws.rd_fh_.mask)
detail::mask_inplace(
*d.fmb, d.ws.rd_key_);
d.fb.commit(bytes_transferred);
d.state = do_control; // VFALCO fall through?
break;
}
assert(d.ws.rd_fh_.op == opcode::close);
{
detail::read(d.ws.cr_, d.fb.data(), code);
if(code != close_code::none)
//------------------------------------------------------------------
case do_control:
if(d.ws.rd_fh_.op == opcode::ping)
{
d.state = 18;
break;
}
if(! d.ws.wr_close_)
{
auto cr = d.ws.cr_;
if(cr.code == close_code::none)
cr.code = close_code::normal;
cr.reason = "";
ping_data data;
detail::read(data, d.fb.data());
d.fb.reset();
d.ws.template write_close<
static_streambuf>(d.fb, cr);
if(d.ws.wr_close_)
{
// ignore ping when closing
d.state = do_read_fh;
break;
}
d.ws.template write_ping<static_streambuf>(
d.fb, opcode::pong, data);
if(d.ws.wr_block_)
{
// suspend
d.state = 9;
d.state = do_pong_resume;
assert(d.ws.wr_block_ != &d);
d.ws.rd_op_.template emplace<
read_frame_op>(std::move(*this));
return;
}
d.state = 11;
d.state = do_pong;
break;
}
// call handler;
d.state = 99;
ec = error::closed;
break;
}
else if(d.ws.rd_fh_.op == opcode::pong)
{
code = close_code::none;
ping_data payload;
detail::read(payload, d.fb.data());
if(d.ws.pong_cb_)
d.ws.pong_cb_(payload);
d.fb.reset();
d.state = do_read_fh;
break;
}
assert(d.ws.rd_fh_.op == opcode::close);
{
detail::read(d.ws.cr_, d.fb.data(), code);
if(code != close_code::none)
{
// protocol error
d.state = do_fail;
break;
}
if(! d.ws.wr_close_)
{
auto cr = d.ws.cr_;
if(cr.code == close_code::none)
cr.code = close_code::normal;
cr.reason = "";
d.fb.reset();
d.ws.template write_close<
static_streambuf>(d.fb, cr);
if(d.ws.wr_block_)
{
// suspend
d.state = do_close_resume;
d.ws.rd_op_.template emplace<
read_frame_op>(std::move(*this));
return;
}
d.state = do_close;
break;
}
// call handler;
ec = error::closed;
goto upcall;
}
// resume
case 9:
d.state = 10;
d.ws.get_io_service().post(bind_handler(
std::move(*this), ec, bytes_transferred));
return;
//------------------------------------------------------------------
case 10:
if(d.ws.error_)
{
// call handler
d.state = 99;
ec = boost::asio::error::operation_aborted;
break;
}
if(d.ws.wr_close_)
{
// call handler
d.state = 99;
ec = error::closed;
break;
}
d.state = 11;
break;
case do_pong_resume:
d.state = do_pong_resume + 1;
d.ws.get_io_service().post(bind_handler(
std::move(*this), ec, bytes_transferred));
return;
// send close
case 11:
d.state = 12;
assert(! d.ws.wr_block_);
d.ws.wr_block_ = &d;
boost::asio::async_write(d.ws.stream_,
d.fb.data(), std::move(*this));
return;;
case do_pong_resume + 1:
if(d.ws.failed_)
{
// call handler
ec = boost::asio::error::operation_aborted;
goto upcall;
}
d.state = do_pong;
break; // VFALCO fall through?
// teardown
case 12:
d.state = 13;
websocket_helpers::call_async_teardown(
d.ws.next_layer(), std::move(*this));
return;
//------------------------------------------------------------------
case 13:
// call handler
d.state = 99;
ec = error::closed;
break;
case do_pong:
if(d.ws.wr_close_)
{
// ignore ping when closing
d.fb.reset();
d.state = do_read_fh;
break;
}
// send pong
d.state = do_pong + 1;
assert(! d.ws.wr_block_);
d.ws.wr_block_ = &d;
boost::asio::async_write(d.ws.stream_,
d.fb.data(), std::move(*this));
return;
// resume
case 14:
d.state = 15;
d.ws.get_io_service().post(bind_handler(
std::move(*this), ec, bytes_transferred));
return;
case 15:
if(d.ws.error_)
{
// call handler
d.state = 99;
ec = boost::asio::error::operation_aborted;
break;
}
if(d.ws.wr_close_)
{
case do_pong + 1:
d.fb.reset();
d.state = 2;
d.state = do_read_fh;
d.ws.wr_block_ = nullptr;
break;
}
d.state = 16;
break;
case 16:
// write ping/pong
d.state = 17;
assert(! d.ws.wr_block_);
d.ws.wr_block_ = &d;
boost::asio::async_write(d.ws.stream_,
d.fb.data(), std::move(*this));
return;
//------------------------------------------------------------------
// sent ping/pong
case 17:
d.fb.reset();
d.state = 2;
d.ws.wr_block_ = nullptr;
break;
case do_close_resume:
d.state = do_close_resume + 1;
d.ws.get_io_service().post(bind_handler(
std::move(*this), ec, bytes_transferred));
return;
// fail the connection
case 18:
if(! d.ws.wr_close_)
{
case do_close_resume + 1:
if(d.ws.failed_)
{
// call handler
d.state = do_call_handler;
ec = boost::asio::error::operation_aborted;
break;
}
if(d.ws.wr_close_)
{
// call handler
ec = error::closed;
goto upcall;
}
d.state = do_close;
break;
//------------------------------------------------------------------
case do_close:
d.state = do_close + 1;
d.ws.wr_close_ = true;
assert(! d.ws.wr_block_);
d.ws.wr_block_ = &d;
boost::asio::async_write(d.ws.stream_,
d.fb.data(), std::move(*this));
return;
case do_close + 1:
d.state = do_close + 2;
websocket_helpers::call_async_teardown(
d.ws.next_layer(), std::move(*this));
return;
case do_close + 2:
// call handler
ec = error::closed;
goto upcall;
//------------------------------------------------------------------
case do_fail:
if(d.ws.wr_close_)
{
d.state = do_fail + 4;
break;
}
d.fb.reset();
d.ws.template write_close<
static_streambuf>(d.fb, code);
if(d.ws.wr_block_)
{
// suspend
d.state = 19;
d.state = do_fail + 2;
d.ws.rd_op_.template emplace<
read_frame_op>(std::move(*this));
return;
}
d.state = 21;
// fall through
case do_fail + 1:
d.ws.failed_ = true;
// send close frame
d.state = do_fail + 4;
d.ws.wr_close_ = true;
assert(! d.ws.wr_block_);
d.ws.wr_block_ = &d;
boost::asio::async_write(d.ws.stream_,
d.fb.data(), std::move(*this));
return;
case do_fail + 2:
d.state = do_fail + 3;
d.ws.get_io_service().post(bind_handler(
std::move(*this), ec, bytes_transferred));
return;
case do_fail + 3:
if(d.ws.failed_)
{
d.state = do_fail + 5;
break;
}
d.state = do_fail + 1;
break;
case do_fail + 4:
d.state = do_fail + 5;
websocket_helpers::call_async_teardown(
d.ws.next_layer(), std::move(*this));
return;
case do_fail + 5:
// call handler
ec = error::failed;
goto upcall;
//------------------------------------------------------------------
case do_call_handler:
goto upcall;
}
d.state = 22;
break;
// resume
case 19:
d.state = 20;
d.ws.get_io_service().post(bind_handler(
std::move(*this), ec, bytes_transferred));
return;
case 20:
if(d.ws.wr_close_)
{
d.state = 22;
break;
}
d.state = 21;
break;
case 21:
// send close
d.state = 22;
d.ws.wr_close_ = true;
assert(! d.ws.wr_block_);
d.ws.wr_block_ = &d;
boost::asio::async_write(d.ws.stream_,
d.fb.data(), std::move(*this));
return;
// teardown
case 22:
d.state = 23;
websocket_helpers::call_async_teardown(
d.ws.next_layer(), std::move(*this));
return;
case 23:
// call handler
d.state = 99;
ec = error::failed;
break;
}
while(! ec);
}
if(ec)
d.ws.error_ = true;
upcall:
if(d.ws.wr_block_ == &d)
d.ws.wr_block_ = nullptr;
d.ws.wr_op_.maybe_invoke();

14
include/beast/websocket/impl/read_op.ipp
View File

@@ -105,24 +105,34 @@ operator()(error_code const& ec, bool again)
{
auto& d = *d_;
d.cont = d.cont || again;
while(! ec && d.state != 99)
while(! ec)
{
switch(d.state)
{
case 0:
// read payload
d.state = 1;
#if 0
// VFALCO This causes dereference of null, because
// the handler is moved from the data block
// before asio_handler_deallocate is called.
d.ws.async_read_frame(
d.fi, d.sb, std::move(*this));
#else
d.ws.async_read_frame(d.fi, d.sb, *this);
#endif
return;
// got payload
case 1:
d.op = d.fi.op;
d.state = d.fi.fin ? 99 : 0;
if(d.fi.fin)
goto upcall;
d.state = 0;
break;
}
}
upcall:
d.h(ec);
}

5
include/beast/websocket/impl/response_op.ipp
View File

@@ -44,6 +44,9 @@ class stream<NextLayer>::response_op
, h(std::forward<DeducedHandler>(h_))
, cont(cont_)
{
// can't call stream::reset() here
// otherwise accept_op will malfunction
//
if(resp.status != 101)
final_ec = error::handshake_failed;
}
@@ -123,7 +126,7 @@ operator()(error_code ec, bool again)
d.state = 99;
ec = d.final_ec;
if(! ec)
d.ws.role_ = role_type::server;
d.ws.open(detail::role_type::server);
break;
}
}

203
include/beast/websocket/impl/stream.ipp
View File

@@ -13,6 +13,7 @@
#include <beast/websocket/impl/accept_op.ipp>
#include <beast/websocket/impl/close_op.ipp>
#include <beast/websocket/impl/handshake_op.ipp>
#include <beast/websocket/impl/ping_op.ipp>
#include <beast/websocket/impl/read_op.ipp>
#include <beast/websocket/impl/read_frame_op.ipp>
#include <beast/websocket/impl/response_op.ipp>
@@ -40,6 +41,13 @@ namespace websocket {
namespace detail {
template<class _>
void
stream_base::open(role_type role)
{
role_ = role;
}
template<class _>
void
stream_base::prepare_fh(close_code::value& code)
@@ -76,7 +84,7 @@ stream_base::prepare_fh(close_code::value& code)
}
rd_size_ += rd_fh_.len;
}
if(rd_size_ > rd_msg_max_)
if(rd_msg_max_ && rd_size_ > rd_msg_max_)
{
code = close_code::too_big;
return;
@@ -100,7 +108,7 @@ stream_base::write_close(
fh.rsv3 = false;
fh.len = cr.code == close_code::none ?
0 : 2 + cr.reason.size();
fh.mask = role_ == role_type::client;
fh.mask = role_ == detail::role_type::client;
if(fh.mask)
fh.key = maskgen_();
detail::write(sb, fh);
@@ -136,7 +144,7 @@ stream_base::write_close(
template<class Streambuf>
void
stream_base::write_ping(Streambuf& sb,
opcode op, ping_payload_type const& data)
opcode op, ping_data const& data)
{
frame_header fh;
fh.op = op;
@@ -184,7 +192,8 @@ accept()
"SyncStream requirements not met");
error_code ec;
accept(boost::asio::null_buffers{}, ec);
detail::maybe_throw(ec, "accept");
if(ec)
throw system_error{ec};
}
template<class NextLayer>
@@ -223,7 +232,8 @@ accept(ConstBufferSequence const& buffers)
"ConstBufferSequence requirements not met");
error_code ec;
accept(buffers, ec);
detail::maybe_throw(ec, "accept");
if(ec)
throw system_error{ec};
}
template<class NextLayer>
@@ -239,6 +249,7 @@ accept(ConstBufferSequence const& buffers, error_code& ec)
"ConstBufferSequence requirements not met");
using boost::asio::buffer_copy;
using boost::asio::buffer_size;
reset();
stream_.buffer().commit(buffer_copy(
stream_.buffer().prepare(
buffer_size(buffers)), buffers));
@@ -279,7 +290,8 @@ accept(http::request_v1<Body, Headers> const& request)
"SyncStream requirements not met");
error_code ec;
accept(request, ec);
detail::maybe_throw(ec, "accept");
if(ec)
throw system_error{ec};
}
template<class NextLayer>
@@ -291,8 +303,11 @@ accept(http::request_v1<Body, Headers> const& req,
{
static_assert(is_SyncStream<next_layer_type>::value,
"SyncStream requirements not met");
reset();
auto const res = build_response(req);
http::write(stream_, res, ec);
if(ec)
return;
if(res.status != 101)
{
ec = error::handshake_failed;
@@ -300,7 +315,7 @@ accept(http::request_v1<Body, Headers> const& req,
// teardown if Connection: close.
return;
}
role_ = role_type::server;
open(detail::role_type::server);
}
template<class NextLayer>
@@ -316,6 +331,7 @@ async_accept(http::request_v1<Body, Headers> const& req,
beast::async_completion<
AcceptHandler, void(error_code)
> completion(handler);
reset();
response_op<decltype(completion.handler)>{
completion.handler, *this, req,
boost_asio_handler_cont_helpers::
@@ -333,7 +349,8 @@ handshake(boost::string_ref const& host,
"SyncStream requirements not met");
error_code ec;
handshake(host, resource, ec);
detail::maybe_throw(ec, "upgrade");
if(ec)
throw system_error{ec};
}
template<class NextLayer>
@@ -344,6 +361,7 @@ handshake(boost::string_ref const& host,
{
static_assert(is_SyncStream<next_layer_type>::value,
"SyncStream requirements not met");
reset();
std::string key;
http::write(stream_,
build_request(host, resource, key), ec);
@@ -383,7 +401,8 @@ close(close_reason const& cr)
"SyncStream requirements not met");
error_code ec;
close(cr, ec);
detail::maybe_throw(ec, "close");
if(ec)
throw system_error{ec};
}
template<class NextLayer>
@@ -398,7 +417,7 @@ close(close_reason const& cr, error_code& ec)
detail::frame_streambuf fb;
write_close<static_streambuf>(fb, cr);
boost::asio::write(stream_, fb.data(), ec);
error_ = ec != 0;
failed_ = ec != 0;
}
template<class NextLayer>
@@ -418,6 +437,45 @@ async_close(close_reason const& cr, CloseHandler&& handler)
return completion.result.get();
}
template<class NextLayer>
void
stream<NextLayer>::
ping(ping_data const& payload)
{
error_code ec;
ping(payload, ec);
if(ec)
throw system_error{ec};
}
template<class NextLayer>
void
stream<NextLayer>::
ping(ping_data const& payload, error_code& ec)
{
detail::frame_streambuf sb;
write_ping<static_streambuf>(
sb, opcode::ping, payload);
boost::asio::write(stream_, sb.data(), ec);
}
template<class NextLayer>
template<class PingHandler>
typename async_completion<
PingHandler, void(error_code)>::result_type
stream<NextLayer>::
async_ping(ping_data const& payload, PingHandler&& handler)
{
static_assert(is_AsyncStream<next_layer_type>::value,
"AsyncStream requirements requirements not met");
beast::async_completion<
PingHandler, void(error_code)
> completion(handler);
ping_op<decltype(completion.handler)>{
completion.handler, *this, payload};
return completion.result.get();
}
template<class NextLayer>
template<class Streambuf>
void
@@ -428,7 +486,8 @@ read(opcode& op, Streambuf& streambuf)
"SyncStream requirements not met");
error_code ec;
read(op, streambuf, ec);
detail::maybe_throw(ec, "read");
if(ec)
throw system_error{ec};
}
template<class NextLayer>
@@ -481,7 +540,8 @@ read_frame(frame_info& fi, Streambuf& streambuf)
"SyncStream requirements not met");
error_code ec;
read_frame(fi, streambuf, ec);
detail::maybe_throw(ec, "read_some");
if(ec)
throw system_error{ec};
}
template<class NextLayer>
@@ -500,8 +560,8 @@ read_frame(frame_info& fi, Streambuf& streambuf, error_code& ec)
// read header
detail::frame_streambuf fb;
do_read_fh(fb, code, ec);
error_ = ec != 0;
if(error_)
failed_ = ec != 0;
if(failed_)
return;
if(code != close_code::none)
break;
@@ -513,8 +573,8 @@ read_frame(frame_info& fi, Streambuf& streambuf, error_code& ec)
auto const mb = fb.prepare(
static_cast<std::size_t>(rd_fh_.len));
fb.commit(boost::asio::read(stream_, mb, ec));
error_ = ec != 0;
if(error_)
failed_ = ec != 0;
if(failed_)
return;
if(rd_fh_.mask)
detail::mask_inplace(mb, rd_key_);
@@ -522,27 +582,23 @@ read_frame(frame_info& fi, Streambuf& streambuf, error_code& ec)
}
if(rd_fh_.op == opcode::ping)
{
ping_payload_type data;
detail::read(data, fb.data(), code);
if(code != close_code::none)
break;
ping_data data;
detail::read(data, fb.data());
fb.reset();
write_ping<static_streambuf>(
fb, opcode::pong, data);
boost::asio::write(stream_, fb.data(), ec);
error_ = ec != 0;
if(error_)
failed_ = ec != 0;
if(failed_)
return;
continue;
}
else if(rd_fh_.op == opcode::pong)
{
ping_payload_type data;
detail::read(data, fb.data(), code);
if(code != close_code::none)
break;
// VFALCO How to notify callers using
// the synchronous interface?
ping_data payload;
detail::read(payload, fb.data());
if(pong_cb_)
pong_cb_(payload);
continue;
}
assert(rd_fh_.op == opcode::close);
@@ -560,8 +616,8 @@ read_frame(frame_info& fi, Streambuf& streambuf, error_code& ec)
wr_close_ = true;
write_close<static_streambuf>(fb, cr);
boost::asio::write(stream_, fb.data(), ec);
error_ = ec != 0;
if(error_)
failed_ = ec != 0;
if(failed_)
return;
}
break;
@@ -578,8 +634,8 @@ read_frame(frame_info& fi, Streambuf& streambuf, error_code& ec)
detail::clamp(rd_need_));
auto const bytes_transferred =
stream_.read_some(smb, ec);
error_ = ec != 0;
if(error_)
failed_ = ec != 0;
if(failed_)
return;
rd_need_ -= bytes_transferred;
auto const pb = prepare_buffers(
@@ -610,23 +666,23 @@ read_frame(frame_info& fi, Streambuf& streambuf, error_code& ec)
detail::frame_streambuf fb;
write_close<static_streambuf>(fb, code);
boost::asio::write(stream_, fb.data(), ec);
error_ = ec != 0;
if(error_)
failed_ = ec != 0;
if(failed_)
return;
}
websocket_helpers::call_teardown(next_layer(), ec);
error_ = ec != 0;
if(error_)
failed_ = ec != 0;
if(failed_)
return;
ec = error::failed;
error_ = true;
failed_ = true;
return;
}
if(! ec)
websocket_helpers::call_teardown(next_layer(), ec);
if(! ec)
ec = error::closed;
error_ = ec != 0;
failed_ = ec != 0;
}
template<class NextLayer>
@@ -658,7 +714,8 @@ write(ConstBufferSequence const& buffers)
"SyncStream requirements not met");
error_code ec;
write(buffers, ec);
detail::maybe_throw(ec, "write");
if(ec)
throw system_error{ec};
}
template<class NextLayer>
@@ -719,7 +776,8 @@ write_frame(bool fin, ConstBufferSequence const& buffers)
"SyncStream requirements not met");
error_code ec;
write_frame(fin, buffers, ec);
detail::maybe_throw(ec, "write");
if(ec)
throw system_error{ec};
}
template<class NextLayer>
@@ -744,7 +802,7 @@ write_frame(bool fin, ConstBufferSequence const& bs, error_code& ec)
fh.rsv2 = false;
fh.rsv3 = false;
fh.len = buffer_size(bs);
fh.mask = role_ == role_type::client;
fh.mask = role_ == detail::role_type::client;
if(fh.mask)
fh.key = maskgen_();
detail::fh_streambuf fh_buf;
@@ -754,22 +812,21 @@ write_frame(bool fin, ConstBufferSequence const& bs, error_code& ec)
// send header and payload
boost::asio::write(stream_,
buffer_cat(fh_buf.data(), bs), ec);
error_ = ec != 0;
failed_ = ec != 0;
return;
}
detail::prepared_key_type key;
detail::prepare_key(key, fh.key);
auto const tmp_size = detail::clamp(
fh.len, wr_buf_size_);
auto const tmp_size =
detail::clamp(fh.len, mask_buf_size_);
std::unique_ptr<std::uint8_t[]> up(
new std::uint8_t[tmp_size]);
auto const tmp = up.get();
std::uint64_t remain = fh.len;
consuming_buffers<ConstBufferSequence> cb(bs);
{
auto const n =
detail::clamp(remain, tmp_size);
mutable_buffers_1 mb{tmp, n};
mutable_buffers_1 mb{up.get(), n};
buffer_copy(mb, cb);
cb.consume(n);
remain -= n;
@@ -779,7 +836,7 @@ write_frame(bool fin, ConstBufferSequence const& bs, error_code& ec)
buffer_cat(fh_buf.data(), mb), ec);
if(ec)
{
error_ = ec != 0;
failed_ = ec != 0;
return;
}
}
@@ -787,7 +844,7 @@ write_frame(bool fin, ConstBufferSequence const& bs, error_code& ec)
{
auto const n =
detail::clamp(remain, tmp_size);
mutable_buffers_1 mb{tmp, n};
mutable_buffers_1 mb{up.get(), n};
buffer_copy(mb, cb);
cb.consume(n);
remain -= n;
@@ -796,7 +853,7 @@ write_frame(bool fin, ConstBufferSequence const& bs, error_code& ec)
boost::asio::write(stream_, mb, ec);
if(ec)
{
error_ = ec != 0;
failed_ = ec != 0;
return;
}
}
@@ -826,6 +883,23 @@ async_write_frame(bool fin,
//------------------------------------------------------------------------------
template<class NextLayer>
void
stream<NextLayer>::
reset()
{
failed_ = false;
rd_need_ = 0;
rd_cont_ = false;
wr_close_ = false;
wr_cont_ = false;
wr_block_ = nullptr; // should be nullptr on close anyway
pong_data_ = nullptr; // should be nullptr on close anyway
stream_.buffer().consume(
stream_.buffer().size());
}
template<class NextLayer>
http::request_v1<http::empty_body>
stream<NextLayer>::
@@ -860,8 +934,11 @@ build_response(http::request_v1<Body, Headers> const& req)
res.reason = http::reason_string(res.status);
res.version = req.version;
res.body = text;
// VFALCO TODO respect keep-alive here
prepare(res);
(*d_)(res);
prepare(res,
(is_keep_alive(req) && keep_alive_) ?
http::connection::keep_alive :
http::connection::close);
return res;
};
if(req.version < 11)
@@ -874,17 +951,28 @@ build_response(http::request_v1<Body, Headers> const& req)
return err("Missing Host");
if(! req.headers.exists("Sec-WebSocket-Key"))
return err("Missing Sec-WebSocket-Key");
if(! rfc2616::token_in_list(
req.headers["Upgrade"], "websocket"))
return err("Missing websocket Upgrade token");
{
auto const version =
req.headers["Sec-WebSocket-Version"];
if(version.empty())
return err("Missing Sec-WebSocket-Version");
if(version != "13")
return err("Unsupported Sec-WebSocket-Version");
{
http::response_v1<http::string_body> res;
res.status = 426;
res.reason = http::reason_string(res.status);
res.version = req.version;
res.headers.insert("Sec-WebSocket-Version", "13");
prepare(res,
(is_keep_alive(req) && keep_alive_) ?
http::connection::keep_alive :
http::connection::close);
return res;
}
}
if(! rfc2616::token_in_list(
req.headers["Upgrade"], "websocket"))
return err("Missing websocket Upgrade token");
http::response_v1<http::string_body> res;
res.status = 101;
res.reason = http::reason_string(res.status);
@@ -893,7 +981,6 @@ build_response(http::request_v1<Body, Headers> const& req)
{
auto const key =
req.headers["Sec-WebSocket-Key"];
res.headers.insert("Sec-WebSocket-Key", key);
res.headers.insert("Sec-WebSocket-Accept",
detail::make_sec_ws_accept(key));
}
@@ -912,6 +999,8 @@ do_response(http::response_v1<Body, Headers> const& res,
{
// VFALCO Review these error codes
auto fail = [&]{ ec = error::response_failed; };
if(res.version < 11)
return fail();
if(res.status != 101)
return fail();
if(! is_upgrade(res))
@@ -924,7 +1013,7 @@ do_response(http::response_v1<Body, Headers> const& res,
if(res.headers["Sec-WebSocket-Accept"] !=
detail::make_sec_ws_accept(key))
return fail();
role_ = role_type::client;
open(detail::role_type::client);
}
template<class NextLayer>

96
include/beast/websocket/impl/write_frame_op.ipp
View File

@@ -61,13 +61,13 @@ class stream<NextLayer>::write_frame_op
fh.rsv2 = false;
fh.rsv3 = false;
fh.len = boost::asio::buffer_size(cb);
fh.mask = ws.role_ == role_type::client;
fh.mask = ws.role_ == detail::role_type::client;
if(fh.mask)
{
fh.key = ws.maskgen_();
detail::prepare_key(key, fh.key);
tmp_size = detail::clamp(
fh.len, ws.wr_buf_size_);
fh.len, ws.mask_buf_size_);
tmp = boost_asio_handler_alloc_helpers::
allocate(tmp_size, h);
remain = fh.len;
@@ -100,16 +100,17 @@ public:
std::forward<DeducedHandler>(h), ws,
std::forward<Args>(args)...))
{
(*this)(error_code{}, 0, false);
(*this)(error_code{}, false);
}
void operator()()
{
(*this)(error_code{}, 0, true);
(*this)(error_code{});
}
void operator()(error_code ec,
std::size_t bytes_transferred, bool again = true);
void operator()(error_code ec, std::size_t);
void operator()(error_code ec, bool again = true);
friend
void* asio_handler_allocate(
@@ -142,19 +143,33 @@ public:
}
};
template<class NextLayer>
template<class Buffers, class Handler>
void
stream<NextLayer>::
write_frame_op<Buffers, Handler>::
operator()(error_code ec, std::size_t)
{
auto& d = *d_;
if(ec)
d.ws.failed_ = true;
(*this)(ec);
}
template<class NextLayer>
template<class Buffers, class Handler>
void
stream<NextLayer>::
write_frame_op<Buffers, Handler>::
operator()(
error_code ec, std::size_t bytes_transferred, bool again)
operator()(error_code ec, bool again)
{
using boost::asio::buffer_copy;
using boost::asio::mutable_buffers_1;
auto& d = *d_;
d.cont = d.cont || again;
while(! ec && d.state != 99)
if(ec)
goto upcall;
for(;;)
{
switch(d.state)
{
@@ -162,47 +177,27 @@ operator()(
if(d.ws.wr_block_)
{
// suspend
d.state = 1;
d.state = 3;
d.ws.wr_op_.template emplace<
write_frame_op>(std::move(*this));
return;
}
if(d.ws.error_)
if(d.ws.failed_ || d.ws.wr_close_)
{
// call handler
d.state = 99;
d.ws.get_io_service().post(
bind_handler(std::move(*this),
boost::asio::error::operation_aborted, 0));
boost::asio::error::operation_aborted));
return;
}
assert(! d.ws.wr_close_);
d.state = 3;
break;
// fall through
// resume
case 1:
d.state = 2;
d.ws.get_io_service().post(bind_handler(
std::move(*this), ec, bytes_transferred));
return;
case 2:
if(d.ws.error_)
{
// call handler
d.state = 99;
ec = boost::asio::error::operation_aborted;
break;
}
d.state = 3;
break;
case 3:
{
if(! d.fh.mask)
{
// send header and payload
// send header and entire payload
d.state = 99;
assert(! d.ws.wr_block_);
d.ws.wr_block_ = &d;
@@ -219,7 +214,7 @@ operator()(
d.remain -= n;
detail::mask_inplace(mb, d.key);
// send header and payload
d.state = d.remain > 0 ? 4 : 99;
d.state = d.remain > 0 ? 2 : 99;
assert(! d.ws.wr_block_);
d.ws.wr_block_ = &d;
boost::asio::async_write(d.ws.stream_,
@@ -229,7 +224,7 @@ operator()(
}
// sent masked payload
case 4:
case 2:
{
auto const n =
detail::clamp(d.remain, d.tmp_size);
@@ -242,24 +237,41 @@ operator()(
// send payload
if(d.remain == 0)
d.state = 99;
assert(! d.ws.wr_block_);
d.ws.wr_block_ = &d;
assert(d.ws.wr_block_ == &d);
boost::asio::async_write(
d.ws.stream_, mb, std::move(*this));
return;
}
case 3:
d.state = 4;
d.ws.get_io_service().post(bind_handler(
std::move(*this), ec));
return;
case 4:
if(d.ws.failed_ || d.ws.wr_close_)
{
// call handler
ec = boost::asio::error::operation_aborted;
goto upcall;
}
d.state = 1;
break;
case 99:
goto upcall;
}
}
if(ec)
d.ws.error_ = true;
if(d.ws.wr_block_ == &d)
d.ws.wr_block_ = nullptr;
upcall:
if(d.tmp)
{
boost_asio_handler_alloc_helpers::
deallocate(d.tmp, d.tmp_size, d.h);
d.tmp = nullptr;
}
if(d.ws.wr_block_ == &d)
d.ws.wr_block_ = nullptr;
d.ws.rd_op_.maybe_invoke();
d.h(ec);
}

5
include/beast/websocket/impl/write_op.ipp
View File

@@ -109,7 +109,7 @@ operator()(error_code ec, bool again)
{
auto& d = *d_;
d.cont = d.cont || again;
while(! ec && d.state != 99)
if(! ec)
{
switch(d.state)
{
@@ -126,6 +126,9 @@ operator()(error_code ec, bool again)
d.ws.async_write_frame(fin, pb, std::move(*this));
return;
}
case 99:
break;
}
}
d.h(ec);