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

Add permessage-deflate WebSocket extension:

Browse Source 
This implements the permessage-deflate WebSocket
extension as described in HyBi Working Group
draft-ietf-hybi-permessage-compression-28:
https://tools.ietf.org/html/draft-ietf-hybi-permessage-compression-28

This extension allows messages to be compressed using
the raw "deflate" algorithm described in RFC 1951,
"DEFLATE Compressed Data Format Specification version 1.3":
https://www.ietf.org/rfc/rfc1951.txt
This commit is contained in:
Vinnie Falco committed 2017-07-20 08:12:14 -07:00
1 parent 99706347cd
commit 911617c43f
26 files changed
+2404 -1324
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

include/beast/websocket/impl/accept.ipp
+9 -4
View File
@@ -35,7 +35,7 @@ class stream<NextLayer>::response_op
{
bool cont;
stream<NextLayer>& ws;
http::response<http::string_body> resp;
http::response<http::string_body> res;
error_code final_ec;
int state = 0;
@@ -45,12 +45,12 @@ class stream<NextLayer>::response_op
bool cont_)
: cont(cont_)
, ws(ws_)
, resp(ws_.build_response(req))
, res(ws_.build_response(req))
{
// can't call stream::reset() here
// otherwise accept_op will malfunction
//
if(resp.status != 101)
if(res.status != 101)
final_ec = error::handshake_failed;
}
};
@@ -121,7 +121,7 @@ operator()(error_code ec, bool again)
// send response
d.state = 1;
http::async_write(d.ws.next_layer(),
d.resp, std::move(*this));
d.res, std::move(*this));
return;
// sent response
@@ -129,7 +129,11 @@ operator()(error_code ec, bool again)
d.state = 99;
ec = d.final_ec;
if(! ec)
{
pmd_read(
d.ws.pmd_config_, d.res.fields);
d.ws.open(detail::role_type::server);
}
break;
}
}
@@ -412,6 +416,7 @@ accept(http::request<Body, Fields> const& req,
// teardown if Connection: close.
return;
}
pmd_read(pmd_config_, req.fields);
open(detail::role_type::server);
}
include/beast/websocket/impl/handshake.ipp
+8 -2
View File
@@ -118,6 +118,8 @@ operator()(error_code ec, bool again)
d.state = 1;
// VFALCO Do we need the ability to move
// a message on the async_write?
pmd_read(
d.ws.pmd_config_, d.req.fields);
http::async_write(d.ws.stream_,
d.req, std::move(*this));
return;
@@ -187,8 +189,12 @@ handshake(boost::string_ref const& host,
"SyncStream requirements not met");
reset();
std::string key;
http::write(stream_,
build_request(host, resource, key), ec);
{
auto const req =
build_request(host, resource, key);
pmd_read(pmd_config_, req.fields);
http::write(stream_, req, ec);
}
if(ec)
return;
http::response<http::string_body> res;
include/beast/websocket/impl/read.ipp
+346 -122
View File
@@ -18,6 +18,7 @@
#include <beast/core/detail/clamp.hpp>
#include <boost/assert.hpp>
#include <boost/optional.hpp>
#include <limits>
#include <memory>
namespace beast {
@@ -48,6 +49,9 @@ class stream<NextLayer>::read_frame_op
frame_info& fi;
DynamicBuffer& db;
fb_type fb;
std::uint64_t remain;
detail::frame_header fh;
detail::prepared_key key;
boost::optional<dmb_type> dmb;
boost::optional<fmb_type> fmb;
int state = 0;
@@ -142,23 +146,26 @@ template<class NextLayer>
template<class DynamicBuffer, class Handler>
void
stream<NextLayer>::read_frame_op<DynamicBuffer, Handler>::
operator()(error_code ec,std::size_t bytes_transferred, bool again)
operator()(error_code ec,
std::size_t bytes_transferred, bool again)
{
using beast::detail::clamp;
using boost::asio::buffer;
enum
{
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_teardown = 16,
do_fail = 18,
do_inflate_payload = 30,
do_frame_done = 4,
do_read_fh = 5,
do_control_payload = 8,
do_control = 9,
do_pong_resume = 10,
do_pong = 12,
do_close_resume = 14,
do_close = 16,
do_teardown = 17,
do_fail = 19,
do_call_handler = 99
};
@@ -181,32 +188,51 @@ operator()(error_code ec,std::size_t bytes_transferred, bool again)
boost::asio::error::operation_aborted, 0));
return;
}
d.state = d.ws.rd_need_ > 0 ?
do_read_payload : do_read_fh;
d.state = do_read_fh;
break;
//------------------------------------------------------------------
case do_read_payload:
d.state = do_read_payload + 1;
d.dmb = d.db.prepare(clamp(d.ws.rd_need_));
// receive payload data
if(d.fh.len == 0)
{
d.state = do_frame_done;
break;
}
// Enforce message size limit
if(d.ws.rd_msg_max_ && d.fh.len >
d.ws.rd_msg_max_ - d.ws.rd_.size)
{
code = close_code::too_big;
d.state = do_fail;
break;
}
d.ws.rd_.size += d.fh.len;
d.remain = d.fh.len;
if(d.fh.mask)
detail::prepare_key(d.key, d.fh.key);
// fall through
case do_read_payload + 1:
d.state = do_read_payload + 2;
d.dmb = d.db.prepare(clamp(d.remain));
// Read frame payload data
d.ws.stream_.async_read_some(
*d.dmb, std::move(*this));
return;
case do_read_payload + 1:
case do_read_payload + 2:
{
d.ws.rd_need_ -= bytes_transferred;
d.remain -= bytes_transferred;
auto const pb = prepare_buffers(
bytes_transferred, *d.dmb);
if(d.ws.rd_fh_.mask)
detail::mask_inplace(pb, d.ws.rd_key_);
if(d.ws.rd_opcode_ == opcode::text)
if(d.fh.mask)
detail::mask_inplace(pb, d.key);
if(d.ws.rd_.op == 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()))
if(! d.ws.rd_.utf8.write(pb) ||
(d.remain == 0 && d.fh.fin &&
! d.ws.rd_.utf8.finish()))
{
// invalid utf8
code = close_code::bad_payload;
@@ -215,21 +241,102 @@ operator()(error_code ec,std::size_t bytes_transferred, bool again)
}
}
d.db.commit(bytes_transferred);
if(d.ws.rd_need_ > 0)
if(d.remain > 0)
{
d.state = do_read_payload;
d.state = do_read_payload + 1;
break;
}
d.state = do_frame_done;
break;
}
//------------------------------------------------------------------
case do_inflate_payload:
d.remain = d.fh.len;
if(d.fh.len == 0)
{
// inflate even if fh.len == 0, otherwise we
// never emit the end-of-stream deflate block.
bytes_transferred = 0;
d.state = do_inflate_payload + 2;
break;
}
if(d.fh.mask)
detail::prepare_key(d.key, d.fh.key);
// fall through
case do_inflate_payload + 1:
{
d.state = do_inflate_payload + 2;
// Read compressed frame payload data
d.ws.stream_.async_read_some(
buffer(d.ws.rd_.buf.get(), clamp(
d.remain, d.ws.rd_.buf_size)),
std::move(*this));
return;
}
case do_inflate_payload + 2:
{
d.remain -= bytes_transferred;
auto const in = buffer(
d.ws.rd_.buf.get(), bytes_transferred);
if(d.fh.mask)
detail::mask_inplace(in, d.key);
auto const prev = d.db.size();
detail::inflate(d.ws.pmd_->zi, d.db, in, ec);
d.ws.failed_ = ec != 0;
if(d.ws.failed_)
break;
if(d.remain == 0 && d.fh.fin)
{
static std::uint8_t constexpr
empty_block[4] = {
0x00, 0x00, 0xff, 0xff };
detail::inflate(d.ws.pmd_->zi, d.db,
buffer(&empty_block[0], 4), ec);
d.ws.failed_ = ec != 0;
if(d.ws.failed_)
break;
}
if(d.ws.rd_.op == opcode::text)
{
consuming_buffers<typename
DynamicBuffer::const_buffers_type
> cb{d.db.data()};
cb.consume(prev);
if(! d.ws.rd_.utf8.write(cb) ||
(d.remain == 0 && d.fh.fin &&
! d.ws.rd_.utf8.finish()))
{
// invalid utf8
code = close_code::bad_payload;
d.state = do_fail;
break;
}
}
if(d.remain > 0)
{
d.state = do_inflate_payload + 1;
break;
}
if(d.fh.fin && (
(d.ws.role_ == detail::role_type::client &&
d.ws.pmd_config_.server_no_context_takeover) ||
(d.ws.role_ == detail::role_type::server &&
d.ws.pmd_config_.client_no_context_takeover)))
d.ws.pmd_->zi.reset();
d.state = do_frame_done;
break;
}
//------------------------------------------------------------------
case do_frame_done:
// call handler
d.fi.op = d.ws.rd_opcode_;
d.fi.fin = d.ws.rd_fh_.fin &&
d.ws.rd_need_ == 0;
d.fi.op = d.ws.rd_.op;
d.fi.fin = d.fh.fin;
goto upcall;
//------------------------------------------------------------------
@@ -244,7 +351,8 @@ operator()(error_code ec,std::size_t bytes_transferred, bool again)
{
d.fb.commit(bytes_transferred);
code = close_code::none;
auto const n = d.ws.read_fh1(d.fb, code);
auto const n = d.ws.read_fh1(
d.fh, d.fb, code);
if(code != close_code::none)
{
// protocol error
@@ -266,21 +374,21 @@ operator()(error_code ec,std::size_t bytes_transferred, bool again)
case do_read_fh + 2:
d.fb.commit(bytes_transferred);
code = close_code::none;
d.ws.read_fh2(d.fb, code);
d.ws.read_fh2(d.fh, d.fb, code);
if(code != close_code::none)
{
// protocol error
d.state = do_fail;
break;
}
if(detail::is_control(d.ws.rd_fh_.op))
if(detail::is_control(d.fh.op))
{
if(d.ws.rd_fh_.len > 0)
if(d.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));
std::size_t>(d.fh.len));
boost::asio::async_read(d.ws.stream_,
*d.fmb, std::move(*this));
return;
@@ -288,21 +396,29 @@ operator()(error_code ec,std::size_t bytes_transferred, bool again)
d.state = do_control;
break;
}
if(d.ws.rd_need_ > 0)
if(d.fh.op == opcode::text ||
d.fh.op == opcode::binary)
d.ws.rd_begin();
if(d.fh.len == 0 && ! d.fh.fin)
{
d.state = do_read_payload;
// Empty message frame
d.state = do_frame_done;
break;
}
// empty frame
d.state = do_frame_done;
if(! d.ws.pmd_ || ! d.ws.pmd_->rd_set)
d.state = do_read_payload;
else
d.state = do_inflate_payload;
break;
//------------------------------------------------------------------
case do_control_payload:
if(d.ws.rd_fh_.mask)
detail::mask_inplace(
*d.fmb, d.ws.rd_key_);
if(d.fh.mask)
{
detail::prepare_key(d.key, d.fh.key);
detail::mask_inplace(*d.fmb, d.key);
}
d.fb.commit(bytes_transferred);
d.state = do_control; // VFALCO fall through?
break;
@@ -310,7 +426,7 @@ operator()(error_code ec,std::size_t bytes_transferred, bool again)
//------------------------------------------------------------------
case do_control:
if(d.ws.rd_fh_.op == opcode::ping)
if(d.fh.op == opcode::ping)
{
ping_data data;
detail::read(data, d.fb.data());
@@ -335,7 +451,7 @@ operator()(error_code ec,std::size_t bytes_transferred, bool again)
d.state = do_pong;
break;
}
else if(d.ws.rd_fh_.op == opcode::pong)
else if(d.fh.op == opcode::pong)
{
code = close_code::none;
ping_data payload;
@@ -346,7 +462,7 @@ operator()(error_code ec,std::size_t bytes_transferred, bool again)
d.state = do_read_fh;
break;
}
BOOST_ASSERT(d.ws.rd_fh_.op == opcode::close);
BOOST_ASSERT(d.fh.op == opcode::close);
{
detail::read(d.ws.cr_, d.fb.data(), code);
if(code != close_code::none)
@@ -589,110 +705,218 @@ read_frame(frame_info& fi, DynamicBuffer& dynabuf, error_code& ec)
static_assert(beast::is_DynamicBuffer<DynamicBuffer>::value,
"DynamicBuffer requirements not met");
using beast::detail::clamp;
using boost::asio::buffer;
using boost::asio::buffer_cast;
using boost::asio::buffer_size;
close_code::value code{};
for(;;)
{
if(rd_need_ == 0)
// Read frame header
detail::frame_header fh;
detail::frame_streambuf fb;
{
// read header
detail::frame_streambuf fb;
do_read_fh(fb, code, ec);
fb.commit(boost::asio::read(
stream_, fb.prepare(2), ec));
failed_ = ec != 0;
if(failed_)
return;
{
auto const n = read_fh1(fh, fb, code);
if(code != close_code::none)
goto do_close;
if(n > 0)
{
fb.commit(boost::asio::read(
stream_, fb.prepare(n), ec));
failed_ = ec != 0;
if(failed_)
return;
}
}
read_fh2(fh, fb, code);
failed_ = ec != 0;
if(failed_)
return;
if(code != close_code::none)
break;
if(detail::is_control(rd_fh_.op))
goto do_close;
}
if(detail::is_control(fh.op))
{
// Read control frame payload
if(fh.len > 0)
{
// read control payload
if(rd_fh_.len > 0)
auto const mb = fb.prepare(
static_cast<std::size_t>(fh.len));
fb.commit(boost::asio::read(stream_, mb, ec));
failed_ = ec != 0;
if(failed_)
return;
if(fh.mask)
{
auto const mb = fb.prepare(
static_cast<std::size_t>(rd_fh_.len));
fb.commit(boost::asio::read(stream_, mb, ec));
failed_ = ec != 0;
if(failed_)
return;
if(rd_fh_.mask)
detail::mask_inplace(mb, rd_key_);
fb.commit(static_cast<std::size_t>(rd_fh_.len));
detail::prepared_key key;
detail::prepare_key(key, fh.key);
detail::mask_inplace(mb, key);
}
if(rd_fh_.op == opcode::ping)
fb.commit(static_cast<std::size_t>(fh.len));
}
// Process control frame
if(fh.op == opcode::ping)
{
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);
failed_ = ec != 0;
if(failed_)
return;
continue;
}
else if(fh.op == opcode::pong)
{
ping_data payload;
detail::read(payload, fb.data());
if(pong_cb_)
pong_cb_(payload);
continue;
}
BOOST_ASSERT(fh.op == opcode::close);
{
detail::read(cr_, fb.data(), code);
if(code != close_code::none)
goto do_close;
if(! wr_close_)
{
ping_data data;
detail::read(data, fb.data());
auto cr = cr_;
if(cr.code == close_code::none)
cr.code = close_code::normal;
cr.reason = "";
fb.reset();
write_ping<static_streambuf>(
fb, opcode::pong, data);
wr_close_ = true;
write_close<static_streambuf>(fb, cr);
boost::asio::write(stream_, fb.data(), ec);
failed_ = ec != 0;
if(failed_)
return;
continue;
}
else if(rd_fh_.op == opcode::pong)
{
ping_data payload;
detail::read(payload, fb.data());
if(pong_cb_)
pong_cb_(payload);
continue;
}
BOOST_ASSERT(rd_fh_.op == opcode::close);
{
detail::read(cr_, fb.data(), code);
if(code != close_code::none)
break;
if(! wr_close_)
{
auto cr = cr_;
if(cr.code == close_code::none)
cr.code = close_code::normal;
cr.reason = "";
fb.reset();
wr_close_ = true;
write_close<static_streambuf>(fb, cr);
boost::asio::write(stream_, fb.data(), ec);
failed_ = ec != 0;
if(failed_)
return;
}
break;
}
}
if(rd_need_ == 0 && ! rd_fh_.fin)
{
// empty frame
continue;
goto do_close;
}
}
// read payload
auto smb = dynabuf.prepare(clamp(rd_need_));
auto const bytes_transferred =
stream_.read_some(smb, ec);
failed_ = ec != 0;
if(failed_)
return;
rd_need_ -= bytes_transferred;
auto const pb = prepare_buffers(
bytes_transferred, smb);
if(rd_fh_.mask)
detail::mask_inplace(pb, rd_key_);
if(rd_opcode_ == opcode::text)
if(fh.op != opcode::cont)
rd_begin();
if(fh.len == 0 && ! fh.fin)
{
if(! rd_utf8_check_.write(pb) ||
(rd_need_ == 0 && rd_fh_.fin &&
! rd_utf8_check_.finish()))
// empty frame
continue;
}
auto remain = fh.len;
detail::prepared_key key;
if(fh.mask)
detail::prepare_key(key, fh.key);
if(! pmd_ || ! pmd_->rd_set)
{
// Enforce message size limit
if(rd_msg_max_ && fh.len >
rd_msg_max_ - rd_.size)
{
code = close_code::bad_payload;
break;
code = close_code::too_big;
goto do_close;
}
rd_.size += fh.len;
// Read message frame payload
while(remain > 0)
{
auto b =
dynabuf.prepare(clamp(remain));
auto const bytes_transferred =
stream_.read_some(b, ec);
failed_ = ec != 0;
if(failed_)
return;
BOOST_ASSERT(bytes_transferred > 0);
remain -= bytes_transferred;
auto const pb = prepare_buffers(
bytes_transferred, b);
if(fh.mask)
detail::mask_inplace(pb, key);
if(rd_.op == opcode::text)
{
if(! rd_.utf8.write(pb) ||
(remain == 0 && fh.fin &&
! rd_.utf8.finish()))
{
code = close_code::bad_payload;
goto do_close;
}
}
dynabuf.commit(bytes_transferred);
}
}
dynabuf.commit(bytes_transferred);
fi.op = rd_opcode_;
fi.fin = rd_fh_.fin && rd_need_ == 0;
else
{
// Read compressed message frame payload:
// inflate even if fh.len == 0, otherwise we
// never emit the end-of-stream deflate block.
for(;;)
{
auto const bytes_transferred =
stream_.read_some(buffer(rd_.buf.get(),
clamp(remain, rd_.buf_size)), ec);
failed_ = ec != 0;
if(failed_)
return;
remain -= bytes_transferred;
auto const in = buffer(
rd_.buf.get(), bytes_transferred);
if(fh.mask)
detail::mask_inplace(in, key);
auto const prev = dynabuf.size();
detail::inflate(pmd_->zi, dynabuf, in, ec);
failed_ = ec != 0;
if(failed_)
return;
if(remain == 0 && fh.fin)
{
static std::uint8_t constexpr
empty_block[4] = {
0x00, 0x00, 0xff, 0xff };
detail::inflate(pmd_->zi, dynabuf,
buffer(&empty_block[0], 4), ec);
failed_ = ec != 0;
if(failed_)
return;
}
if(rd_.op == opcode::text)
{
consuming_buffers<typename
DynamicBuffer::const_buffers_type
> cb{dynabuf.data()};
cb.consume(prev);
if(! rd_.utf8.write(cb) || (
remain == 0 && fh.fin &&
! rd_.utf8.finish()))
{
code = close_code::bad_payload;
goto do_close;
}
}
if(remain == 0)
break;
}
if(fh.fin && (
(role_ == detail::role_type::client &&
pmd_config_.server_no_context_takeover) ||
(role_ == detail::role_type::server &&
pmd_config_.client_no_context_takeover)))
pmd_->zi.reset();
}
fi.op = rd_.op;
fi.fin = fh.fin;
return;
}
do_close:
if(code != close_code::none)
{
// Fail the connection (per rfc6455)
include/beast/websocket/impl/stream.ipp
+56 -25
View File
@@ -10,6 +10,7 @@
#include <beast/websocket/teardown.hpp>
#include <beast/websocket/detail/hybi13.hpp>
#include <beast/websocket/detail/pmd_extension.hpp>
#include <beast/http/read.hpp>
#include <beast/http/write.hpp>
#include <beast/http/reason.hpp>
@@ -25,6 +26,7 @@
#include <boost/endian/buffers.hpp>
#include <algorithm>
#include <memory>
#include <stdexcept>
#include <utility>
namespace beast {
@@ -38,6 +40,30 @@ stream(Args&&... args)
{
}
template<class NextLayer>
void
stream<NextLayer>::
set_option(permessage_deflate const& o)
{
if( o.server_max_window_bits > 15 ||
o.server_max_window_bits < 9)
throw std::invalid_argument{
"invalid server_max_window_bits"};
if( o.client_max_window_bits > 15 ||
o.client_max_window_bits < 9)
throw std::invalid_argument{
"invalid client_max_window_bits"};
if( o.compLevel < 0 ||
o.compLevel > 9)
throw std::invalid_argument{
"invalid compLevel"};
if( o.memLevel < 1 ||
o.memLevel > 9)
throw std::invalid_argument{
"invalid memLevel"};
pmd_opts_ = o;
}
//------------------------------------------------------------------------------
template<class NextLayer>
@@ -46,8 +72,7 @@ stream<NextLayer>::
reset()
{
failed_ = false;
rd_need_ = 0;
rd_cont_ = false;
rd_.cont = false;
wr_close_ = false;
wr_.cont = false;
wr_block_ = nullptr; // should be nullptr on close anyway
@@ -72,6 +97,21 @@ build_request(boost::string_ref const& host,
key = detail::make_sec_ws_key(maskgen_);
req.fields.insert("Sec-WebSocket-Key", key);
req.fields.insert("Sec-WebSocket-Version", "13");
if(pmd_opts_.client_enable)
{
detail::pmd_offer config;
config.accept = true;
config.server_max_window_bits =
pmd_opts_.server_max_window_bits;
config.client_max_window_bits =
pmd_opts_.client_max_window_bits;
config.server_no_context_takeover =
pmd_opts_.server_no_context_takeover;
config.client_no_context_takeover =
pmd_opts_.client_no_context_takeover;
detail::pmd_write(
req.fields, config);
}
d_(req);
http::prepare(req, http::connection::upgrade);
return req;
@@ -122,6 +162,7 @@ build_response(http::request<Body, Fields> const& req)
res.reason = http::reason_string(res.status);
res.version = req.version;
res.fields.insert("Sec-WebSocket-Version", "13");
d_(res);
prepare(res,
(is_keep_alive(req) && keep_alive_) ?
http::connection::keep_alive :
@@ -130,6 +171,13 @@ build_response(http::request<Body, Fields> const& req)
}
}
http::response<http::string_body> res;
{
detail::pmd_offer offer;
detail::pmd_offer unused;
pmd_read(offer, req.fields);
pmd_negotiate(
res.fields, unused, offer, pmd_opts_);
}
res.status = 101;
res.reason = http::reason_string(res.status);
res.version = req.version;
@@ -168,32 +216,15 @@ do_response(http::response<Body, Fields> const& res,
if(res.fields["Sec-WebSocket-Accept"] !=
detail::make_sec_ws_accept(key))
return fail();
detail::pmd_offer offer;
pmd_read(offer, res.fields);
// VFALCO see if offer satisfies pmd_config_,
// return an error if not.
#pragma message("Check offer in do_response")
pmd_config_ = offer; // overwrite for now
open(detail::role_type::client);
}
template<class NextLayer>
void
stream<NextLayer>::
do_read_fh(detail::frame_streambuf& fb,
close_code::value& code, error_code& ec)
{
fb.commit(boost::asio::read(
stream_, fb.prepare(2), ec));
if(ec)
return;
auto const n = read_fh1(fb, code);
if(code != close_code::none)
return;
if(n > 0)
{
fb.commit(boost::asio::read(
stream_, fb.prepare(n), ec));
if(ec)
return;
}
read_fh2(fb, code);
}
} // websocket
} // beast
include/beast/websocket/impl/write.ipp
+458 -230
View File
@@ -23,77 +23,11 @@
#include <algorithm>
#include <memory>
#include <beast/unit_test/dstream.hpp>
namespace beast {
namespace websocket {
/*
template<class ConstBufferSequence>
void
write_frame(bool fin, ConstBufferSequence const& buffer)
Depending on the settings of autofragment role, and compression,
different algorithms are used.
1. autofragment: false
compression: false
In the server role, this will send a single frame in one
system call, by concatenating the frame header and the payload.
In the client role, this will send a single frame in one system
call, using the write buffer to calculate masked data.
2. autofragment: true
compression: false
In the server role, this will send one or more frames in one
system call per sent frame. Each frame is sent by concatenating
the frame header and payload. The size of each sent frame will
not exceed the write buffer size option.
In the client role, this will send one or more frames in one
system call per sent frame, using the write buffer to calculate
masked data. The size of each sent frame will not exceed the
write buffer size option.
3. autofragment: false
compression: true
In the server role, this will...
*/
/*
if(compress)
compress buffers into write_buffer
if(write_buffer_avail == write_buffer_size || fin`)
if(mask)
apply mask to write buffer
write frame header, write_buffer as one frame
else if(auto-fragment)
if(fin || write_buffer_avail + buffers size == write_buffer_size)
if(mask)
append buffers to write buffer
apply mask to write buffer
write frame header, write buffer as one frame
else:
write frame header, write buffer, and buffers as one frame
else:
append buffers to write buffer
else if(mask)
copy buffers to write_buffer
apply mask to write_buffer
write frame header and possibly full write_buffer in a single call
loop:
copy buffers to write_buffer
apply mask to write_buffer
write write_buffer in a single call
else
write frame header, buffers as one frame
*/
//------------------------------------------------------------------------------
template<class NextLayer>
template<class Buffers, class Handler>
class stream<NextLayer>::write_frame_op
@@ -104,53 +38,23 @@ class stream<NextLayer>::write_frame_op
bool cont;
stream<NextLayer>& ws;
consuming_buffers<Buffers> cb;
bool fin;
detail::frame_header fh;
detail::fh_streambuf fh_buf;
detail::prepared_key_type key;
void* tmp;
std::size_t tmp_size;
detail::prepared_key key;
std::uint64_t remain;
int state = 0;
int entry;
data(Handler& handler_, stream<NextLayer>& ws_,
bool fin, Buffers const& bs)
bool fin_, Buffers const& bs)
: handler(handler_)
, cont(beast_asio_helpers::
is_continuation(handler))
, ws(ws_)
, cb(bs)
, fin(fin_)
{
using beast::detail::clamp;
fh.op = ws.wr_.cont ?
opcode::cont : ws.wr_opcode_;
ws.wr_.cont = ! fin;
fh.fin = fin;
fh.rsv1 = false;
fh.rsv2 = false;
fh.rsv3 = false;
fh.len = boost::asio::buffer_size(cb);
fh.mask = ws.role_ == detail::role_type::client;
if(fh.mask)
{
fh.key = ws.maskgen_();
detail::prepare_key(key, fh.key);
tmp_size = clamp(fh.len, ws.wr_buf_size_);
tmp = beast_asio_helpers::
allocate(tmp_size, handler);
remain = fh.len;
}
else
{
tmp = nullptr;
}
detail::write<static_streambuf>(fh_buf, fh);
}
~data()
{
if(tmp)
beast_asio_helpers::
deallocate(tmp, tmp_size, handler);
}
};
@@ -167,17 +71,24 @@ public:
std::forward<DeducedHandler>(h),
ws, std::forward<Args>(args)...))
{
(*this)(error_code{}, false);
(*this)(error_code{}, 0, false);
}
void operator()()
{
(*this)(error_code{});
(*this)(error_code{}, 0, true);
}
void operator()(error_code ec, std::size_t);
void operator()(error_code const& ec)
{
(*this)(ec, 0, true);
}
void operator()(error_code ec, bool again = true);
void operator()(error_code ec,
std::size_t bytes_transferred);
void operator()(error_code ec,
std::size_t bytes_transferred, bool again);
friend
void* asio_handler_allocate(
@@ -215,12 +126,12 @@ template<class Buffers, class Handler>
void
stream<NextLayer>::
write_frame_op<Buffers, Handler>::
operator()(error_code ec, std::size_t)
operator()(error_code ec, std::size_t bytes_transferred)
{
auto& d = *d_;
if(ec)
d.ws.failed_ = true;
(*this)(ec);
(*this)(ec, bytes_transferred, true);
}
template<class NextLayer>
@@ -228,11 +139,24 @@ template<class Buffers, class Handler>
void
stream<NextLayer>::
write_frame_op<Buffers, Handler>::
operator()(error_code ec, bool again)
operator()(error_code ec,
std::size_t bytes_transferred, bool again)
{
using beast::detail::clamp;
using boost::asio::buffer;
using boost::asio::buffer_copy;
using boost::asio::mutable_buffers_1;
using boost::asio::buffer_size;
enum
{
do_init = 0,
do_nomask_nofrag = 20,
do_nomask_frag = 30,
do_mask_nofrag = 40,
do_mask_frag = 50,
do_deflate = 60,
do_maybe_suspend = 80,
do_upcall = 99
};
auto& d = *d_;
d.cont = d.cont || again;
if(ec)
@@ -241,11 +165,299 @@ operator()(error_code ec, bool again)
{
switch(d.state)
{
case 0:
case do_init:
if(! d.ws.wr_.cont)
{
d.ws.wr_begin();
d.fh.rsv1 = d.ws.wr_.compress;
}
else
{
d.fh.rsv1 = false;
}
d.fh.rsv2 = false;
d.fh.rsv3 = false;
d.fh.op = d.ws.wr_.cont ?
opcode::cont : d.ws.wr_opcode_;
d.fh.mask =
d.ws.role_ == detail::role_type::client;
if(d.ws.wr_.compress)
{
d.entry = do_deflate;
}
else if(! d.fh.mask)
{
if(! d.ws.wr_.autofrag)
{
d.entry = do_nomask_nofrag;
}
else
{
BOOST_ASSERT(d.ws.wr_.buf_size != 0);
d.remain = buffer_size(d.cb);
if(d.remain > d.ws.wr_.buf_size)
d.entry = do_nomask_frag;
else
d.entry = do_nomask_nofrag;
}
}
else
{
if(! d.ws.wr_.autofrag)
{
d.entry = do_mask_nofrag;
}
else
{
BOOST_ASSERT(d.ws.wr_.buf_size != 0);
d.remain = buffer_size(d.cb);
if(d.remain > d.ws.wr_.buf_size)
d.entry = do_mask_frag;
else
d.entry = do_mask_nofrag;
}
}
d.state = do_maybe_suspend;
break;
//----------------------------------------------------------------------
case do_nomask_nofrag:
{
d.fh.fin = d.fin;
d.fh.len = buffer_size(d.cb);
detail::write<static_streambuf>(
d.fh_buf, d.fh);
d.ws.wr_.cont = ! d.fin;
// Send frame
d.state = do_upcall;
BOOST_ASSERT(! d.ws.wr_block_);
d.ws.wr_block_ = &d;
boost::asio::async_write(d.ws.stream_,
buffer_cat(d.fh_buf.data(), d.cb),
std::move(*this));
return;
}
//----------------------------------------------------------------------
case do_nomask_frag:
{
auto const n = clamp(
d.remain, d.ws.wr_.buf_size);
d.remain -= n;
d.fh.len = n;
d.fh.fin = d.fin ? d.remain == 0 : false;
detail::write<static_streambuf>(
d.fh_buf, d.fh);
d.ws.wr_.cont = ! d.fin;
// Send frame
d.state = d.remain == 0 ?
do_upcall : do_nomask_frag + 1;
BOOST_ASSERT(! d.ws.wr_block_);
d.ws.wr_block_ = &d;
boost::asio::async_write(d.ws.stream_,
buffer_cat(d.fh_buf.data(),
prepare_buffers(n, d.cb)),
std::move(*this));
return;
}
case do_nomask_frag + 1:
d.cb.consume(
bytes_transferred - d.fh_buf.size());
d.fh_buf.reset();
d.fh.op = opcode::cont;
if(d.ws.wr_block_ == &d)
d.ws.wr_block_ = nullptr;
if(d.ws.rd_op_.maybe_invoke())
{
d.state = do_maybe_suspend;
d.ws.get_io_service().post(
std::move(*this));
return;
}
d.state = d.entry;
break;
//----------------------------------------------------------------------
case do_mask_nofrag:
{
d.remain = buffer_size(d.cb);
d.fh.fin = d.fin;
d.fh.len = d.remain;
d.fh.key = d.ws.maskgen_();
detail::prepare_key(d.key, d.fh.key);
detail::write<static_streambuf>(
d.fh_buf, d.fh);
auto const n =
clamp(d.remain, d.ws.wr_.buf_size);
auto const b =
buffer(d.ws.wr_.buf.get(), n);
buffer_copy(b, d.cb);
detail::mask_inplace(b, d.key);
d.remain -= n;
d.ws.wr_.cont = ! d.fin;
// Send frame header and partial payload
d.state = d.remain == 0 ?
do_upcall : do_mask_nofrag + 1;
BOOST_ASSERT(! d.ws.wr_block_);
d.ws.wr_block_ = &d;
boost::asio::async_write(d.ws.stream_,
buffer_cat(d.fh_buf.data(), b),
std::move(*this));
return;
}
case do_mask_nofrag + 1:
{
d.cb.consume(d.ws.wr_.buf_size);
auto const n =
clamp(d.remain, d.ws.wr_.buf_size);
auto const b =
buffer(d.ws.wr_.buf.get(), n);
buffer_copy(b, d.cb);
detail::mask_inplace(b, d.key);
d.remain -= n;
// Send parial payload
if(d.remain == 0)
d.state = do_upcall;
boost::asio::async_write(
d.ws.stream_, b, std::move(*this));
return;
}
//----------------------------------------------------------------------
case do_mask_frag:
{
auto const n = clamp(
d.remain, d.ws.wr_.buf_size);
d.remain -= n;
d.fh.len = n;
d.fh.key = d.ws.maskgen_();
d.fh.fin = d.fin ? d.remain == 0 : false;
detail::prepare_key(d.key, d.fh.key);
auto const b = buffer(
d.ws.wr_.buf.get(), n);
buffer_copy(b, d.cb);
detail::mask_inplace(b, d.key);
detail::write<static_streambuf>(
d.fh_buf, d.fh);
d.ws.wr_.cont = ! d.fin;
// Send frame
d.state = d.remain == 0 ?
do_upcall : do_mask_frag + 1;
BOOST_ASSERT(! d.ws.wr_block_);
d.ws.wr_block_ = &d;
boost::asio::async_write(d.ws.stream_,
buffer_cat(d.fh_buf.data(), b),
std::move(*this));
return;
}
case do_mask_frag + 1:
d.cb.consume(
bytes_transferred - d.fh_buf.size());
d.fh_buf.reset();
d.fh.op = opcode::cont;
BOOST_ASSERT(d.ws.wr_block_ == &d);
d.ws.wr_block_ = nullptr;
if(d.ws.rd_op_.maybe_invoke())
{
d.state = do_maybe_suspend;
d.ws.get_io_service().post(
std::move(*this));
return;
}
d.state = d.entry;
break;
//----------------------------------------------------------------------
case do_deflate:
{
auto b = buffer(d.ws.wr_.buf.get(),
d.ws.wr_.buf_size);
auto const more = detail::deflate(
d.ws.pmd_->zo, b, d.cb, d.fin, ec);
d.ws.failed_ = ec != 0;
if(d.ws.failed_)
goto upcall;
auto const n = buffer_size(b);
if(n == 0)
{
// The input was consumed, but there
// is no output due to compression
// latency.
BOOST_ASSERT(! d.fin);
BOOST_ASSERT(buffer_size(d.cb) == 0);
// We can skip the dispatch if the
// asynchronous initiation function is
// not on call stack but its hard to
// figure out so be safe and dispatch.
d.state = do_upcall;
d.ws.get_io_service().post(std::move(*this));
return;
}
if(d.fh.mask)
{
d.fh.key = d.ws.maskgen_();
detail::prepared_key key;
detail::prepare_key(key, d.fh.key);
detail::mask_inplace(b, key);
}
d.fh.fin = ! more;
d.fh.len = n;
detail::fh_streambuf fh_buf;
detail::write<static_streambuf>(fh_buf, d.fh);
d.ws.wr_.cont = ! d.fin;
// Send frame
d.state = more ?
do_deflate + 1 : do_deflate + 2;
BOOST_ASSERT(! d.ws.wr_block_);
d.ws.wr_block_ = &d;
boost::asio::async_write(d.ws.stream_,
buffer_cat(fh_buf.data(), b),
std::move(*this));
return;
}
case do_deflate + 1:
d.fh.op = opcode::cont;
d.fh.rsv1 = false;
BOOST_ASSERT(d.ws.wr_block_ == &d);
d.ws.wr_block_ = nullptr;
if(d.ws.rd_op_.maybe_invoke())
{
d.state = do_maybe_suspend;
d.ws.get_io_service().post(
std::move(*this));
return;
}
d.state = d.entry;
break;
case do_deflate + 2:
if(d.fh.fin && (
(d.ws.role_ == detail::role_type::client &&
d.ws.pmd_config_.client_no_context_takeover) ||
(d.ws.role_ == detail::role_type::server &&
d.ws.pmd_config_.server_no_context_takeover)))
d.ws.pmd_->zo.reset();
goto upcall;
//----------------------------------------------------------------------
case do_maybe_suspend:
{
if(d.ws.wr_block_)
{
// suspend
d.state = 3;
d.state = do_maybe_suspend + 1;
d.ws.wr_op_.template emplace<
write_frame_op>(std::move(*this));
return;
@@ -253,79 +465,35 @@ operator()(error_code ec, bool again)
if(d.ws.failed_ || d.ws.wr_close_)
{
// call handler
d.state = 99;
d.state = do_upcall;
d.ws.get_io_service().post(
bind_handler(std::move(*this),
boost::asio::error::operation_aborted));
return;
}
// fall through
case 1:
{
if(! d.fh.mask)
{
// send header and entire payload
d.state = 99;
BOOST_ASSERT(! d.ws.wr_block_);
d.ws.wr_block_ = &d;
boost::asio::async_write(d.ws.stream_,
buffer_cat(d.fh_buf.data(), d.cb),
std::move(*this));
return;
}
auto const n = clamp(d.remain, d.tmp_size);
mutable_buffers_1 mb{d.tmp, n};
buffer_copy(mb, d.cb);
d.cb.consume(n);
d.remain -= n;
detail::mask_inplace(mb, d.key);
// send header and payload
d.state = d.remain > 0 ? 2 : 99;
BOOST_ASSERT(! d.ws.wr_block_);
d.ws.wr_block_ = &d;
boost::asio::async_write(d.ws.stream_,
buffer_cat(d.fh_buf.data(),
mb), std::move(*this));
return;
d.state = d.entry;
break;
}
// sent masked payload
case 2:
{
auto const n = clamp(d.remain, d.tmp_size);
mutable_buffers_1 mb{d.tmp,
static_cast<std::size_t>(n)};
buffer_copy(mb, d.cb);
d.cb.consume(n);
d.remain -= n;
detail::mask_inplace(mb, d.key);
// send payload
if(d.remain == 0)
d.state = 99;
BOOST_ASSERT(d.ws.wr_block_ == &d);
boost::asio::async_write(
d.ws.stream_, mb, std::move(*this));
return;
}
case 3:
d.state = 4;
case do_maybe_suspend + 1:
d.state = do_maybe_suspend + 2;
d.ws.get_io_service().post(bind_handler(
std::move(*this), ec));
return;
case 4:
case do_maybe_suspend + 2:
if(d.ws.failed_ || d.ws.wr_close_)
{
// call handler
ec = boost::asio::error::operation_aborted;
goto upcall;
}
d.state = 1;
d.state = d.entry;
break;
case 99:
//----------------------------------------------------------------------
case do_upcall:
goto upcall;
}
}
@@ -391,120 +559,182 @@ write_frame(bool fin,
using boost::asio::buffer;
using boost::asio::buffer_copy;
using boost::asio::buffer_size;
bool const compress = false;
if(! wr_.cont)
wr_prepare(compress);
detail::frame_header fh;
fh.op = wr_.cont ? opcode::cont : wr_opcode_;
fh.rsv1 = false;
if(! wr_.cont)
{
wr_begin();
fh.rsv1 = wr_.compress;
}
else
{
fh.rsv1 = false;
}
fh.rsv2 = false;
fh.rsv3 = false;
fh.op = wr_.cont ? opcode::cont : wr_opcode_;
fh.mask = role_ == detail::role_type::client;
wr_.cont = ! fin;
auto remain = buffer_size(buffers);
if(compress)
if(wr_.compress)
{
// TODO
}
else if(! fh.mask && ! wr_.autofrag)
{
fh.fin = fin;
fh.len = remain;
detail::fh_streambuf fh_buf;
detail::write<static_streambuf>(fh_buf, fh);
boost::asio::write(stream_,
buffer_cat(fh_buf.data(), buffers), ec);
failed_ = ec != 0;
if(failed_)
return;
return;
}
else if(! fh.mask && wr_.autofrag)
{
BOOST_ASSERT(wr_.size != 0);
consuming_buffers<
ConstBufferSequence> cb(buffers);
ConstBufferSequence> cb{buffers};
for(;;)
{
auto const n = clamp(remain, wr_.size);
fh.len = n;
remain -= n;
fh.fin = fin ? remain == 0 : false;
detail::fh_streambuf fh_buf;
detail::write<static_streambuf>(fh_buf, fh);
boost::asio::write(stream_,
buffer_cat(fh_buf.data(),
prepare_buffers(n, cb)), ec);
auto b = buffer(
wr_.buf.get(), wr_.buf_size);
auto const more = detail::deflate(
pmd_->zo, b, cb, fin, ec);
failed_ = ec != 0;
if(failed_)
return;
if(remain == 0)
auto const n = buffer_size(b);
if(n == 0)
{
// The input was consumed, but there
// is no output due to compression
// latency.
BOOST_ASSERT(! fin);
BOOST_ASSERT(buffer_size(cb) == 0);
fh.fin = false;
break;
}
if(fh.mask)
{
fh.key = maskgen_();
detail::prepared_key key;
detail::prepare_key(key, fh.key);
detail::mask_inplace(b, key);
}
fh.fin = ! more;
fh.len = n;
detail::fh_streambuf fh_buf;
detail::write<static_streambuf>(fh_buf, fh);
wr_.cont = ! fin;
boost::asio::write(stream_,
buffer_cat(fh_buf.data(), b), ec);
failed_ = ec != 0;
if(failed_)
return;
if(! more)
break;
fh.op = opcode::cont;
cb.consume(n);
fh.rsv1 = false;
}
if(fh.fin && (
(role_ == detail::role_type::client &&
pmd_config_.client_no_context_takeover) ||
(role_ == detail::role_type::server &&
pmd_config_.server_no_context_takeover)))
pmd_->zo.reset();
return;
}
if(! fh.mask)
{
if(! wr_.autofrag)
{
// no mask, no autofrag
fh.fin = fin;
fh.len = remain;
detail::fh_streambuf fh_buf;
detail::write<static_streambuf>(fh_buf, fh);
wr_.cont = ! fin;
boost::asio::write(stream_,
buffer_cat(fh_buf.data(), buffers), ec);
failed_ = ec != 0;
if(failed_)
return;
}
else
{
// no mask, autofrag
BOOST_ASSERT(wr_.buf_size != 0);
consuming_buffers<
ConstBufferSequence> cb{buffers};
for(;;)
{
auto const n = clamp(remain, wr_.buf_size);
remain -= n;
fh.len = n;
fh.fin = fin ? remain == 0 : false;
detail::fh_streambuf fh_buf;
detail::write<static_streambuf>(fh_buf, fh);
wr_.cont = ! fin;
boost::asio::write(stream_,
buffer_cat(fh_buf.data(),
prepare_buffers(n, cb)), ec);
failed_ = ec != 0;
if(failed_)
return;
if(remain == 0)
break;
fh.op = opcode::cont;
cb.consume(n);
}
}
return;
}
else if(fh.mask && ! wr_.autofrag)
if(! wr_.autofrag)
{
fh.key = maskgen_();
detail::prepared_key_type key;
detail::prepare_key(key, fh.key);
// mask, no autofrag
fh.fin = fin;
fh.len = remain;
fh.key = maskgen_();
detail::prepared_key key;
detail::prepare_key(key, fh.key);
detail::fh_streambuf fh_buf;
detail::write<static_streambuf>(fh_buf, fh);
consuming_buffers<
ConstBufferSequence> cb(buffers);
ConstBufferSequence> cb{buffers};
{
auto const n = clamp(remain, wr_.size);
auto const mb = buffer(wr_.buf.get(), n);
buffer_copy(mb, cb);
auto const n = clamp(remain, wr_.buf_size);
auto const b = buffer(wr_.buf.get(), n);
buffer_copy(b, cb);
cb.consume(n);
remain -= n;
detail::mask_inplace(mb, key);
detail::mask_inplace(b, key);
wr_.cont = ! fin;
boost::asio::write(stream_,
buffer_cat(fh_buf.data(), mb), ec);
buffer_cat(fh_buf.data(), b), ec);
failed_ = ec != 0;
if(failed_)
return;
}
while(remain > 0)
{
auto const n = clamp(remain, wr_.size);
auto const mb = buffer(wr_.buf.get(), n);
buffer_copy(mb, cb);
auto const n = clamp(remain, wr_.buf_size);
auto const b = buffer(wr_.buf.get(), n);
buffer_copy(b, cb);
cb.consume(n);
remain -= n;
detail::mask_inplace(mb, key);
boost::asio::write(stream_, mb, ec);
detail::mask_inplace(b, key);
boost::asio::write(stream_, b, ec);
failed_ = ec != 0;
if(failed_)
return;
}
return;
}
else if(fh.mask && wr_.autofrag)
{
BOOST_ASSERT(wr_.size != 0);
// mask, autofrag
BOOST_ASSERT(wr_.buf_size != 0);
consuming_buffers<
ConstBufferSequence> cb(buffers);
ConstBufferSequence> cb{buffers};
for(;;)
{
fh.key = maskgen_();
detail::prepared_key_type key;
detail::prepared_key key;
detail::prepare_key(key, fh.key);
auto const n = clamp(remain, wr_.size);
auto const mb = buffer(wr_.buf.get(), n);
buffer_copy(mb, cb);
detail::mask_inplace(mb, key);
auto const n = clamp(remain, wr_.buf_size);
auto const b = buffer(wr_.buf.get(), n);
buffer_copy(b, cb);
detail::mask_inplace(b, key);
fh.len = n;
remain -= n;
fh.fin = fin ? remain == 0 : false;
detail::fh_streambuf fh_buf;
detail::write<static_streambuf>(fh_buf, fh);
boost::asio::write(stream_,
buffer_cat(fh_buf.data(), mb), ec);
buffer_cat(fh_buf.data(), b), ec);
failed_ = ec != 0;
if(failed_)
return;
@@ -675,8 +905,6 @@ write(ConstBufferSequence const& buffers, error_code& ec)
write_frame(true, buffers, ec);
}
//------------------------------------------------------------------------------
} // websocket
} // beast