Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: add websocket permessage-deflate #559

Merged
merged 3 commits into from
Apr 19, 2024
Merged

feat: add websocket permessage-deflate #559

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
2a4fdbf
feat: add websocket permessage-deflate
helintongh Apr 18, 2024
3819115
fxi and some simplify
qicosmos Apr 19, 2024
e001f60
set_ws_deflate
qicosmos Apr 19, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
154 changes: 131 additions & 23 deletions include/cinatra/coro_http_client.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,9 @@
#include "async_simple/Unit.h"
#include "async_simple/coro/FutureAwaiter.h"
#include "async_simple/coro/Lazy.h"
#ifdef CINATRA_ENABLE_GZIP
#include "gzip.hpp"
#endif
#include "cinatra_log_wrapper.hpp"
#include "http_parser.hpp"
#include "multipart.hpp"
Expand Down Expand Up @@ -273,6 +276,12 @@ class coro_http_client : public std::enable_shared_from_this<coro_http_client> {
return std::move(body_);
}

#ifdef CINATRA_ENABLE_GZIP
void set_ws_deflate(bool enable_ws_deflate) {
enable_ws_deflate_ = enable_ws_deflate;
}
#endif

// only make socket connet(or handshake) to the host
async_simple::coro::Lazy<resp_data> connect(std::string uri) {
resp_data data{};
Expand All @@ -298,10 +307,30 @@ class coro_http_client : public std::enable_shared_from_this<coro_http_client> {
}
add_header("Sec-WebSocket-Key", ws_sec_key_);
add_header("Sec-WebSocket-Version", "13");

#ifdef CINATRA_ENABLE_GZIP
if (enable_ws_deflate_)
add_header("Sec-WebSocket-Extensions",
"permessage-deflate; client_max_window_bits");
#endif
req_context<> ctx{};
data = co_await async_request(std::move(uri), http_method::GET,
std::move(ctx));

#ifdef CINATRA_ENABLE_GZIP
if (enable_ws_deflate_) {
for (auto c : data.resp_headers) {
if (c.name == "Sec-WebSocket-Extensions") {
if (c.value.find("permessage-deflate;") != std::string::npos) {
is_server_support_ws_deflate_ = true;
}
else {
is_server_support_ws_deflate_ = false;
}
break;
}
}
}
#endif
co_return data;
}
data = co_await connect(u);
Expand Down Expand Up @@ -382,37 +411,91 @@ class coro_http_client : public std::enable_shared_from_this<coro_http_client> {
}

if constexpr (is_span_v<Source>) {
std::string encode_header = ws.encode_frame(source, op, true);
std::vector<asio::const_buffer> buffers{
asio::buffer(encode_header.data(), encode_header.size()),
asio::buffer(source.data(), source.size())};

auto [ec, _] = co_await async_write(buffers);
if (ec) {
data.net_err = ec;
data.status = 404;
#ifdef CINATRA_ENABLE_GZIP
if (enable_ws_deflate_ && is_server_support_ws_deflate_) {
std::string dest_buf;
if (cinatra::gzip_codec::deflate({source.data(), source.size()},
dest_buf)) {
std::span<char> msg(dest_buf.data(), dest_buf.size());
auto header = ws.encode_frame(msg, op, true, true);
std::vector<asio::const_buffer> buffers{asio::buffer(header),
asio::buffer(dest_buf)};

auto [ec, sz] = co_await async_write(buffers);
if (ec) {
data.net_err = ec;
data.status = 404;
}
}
else {
CINATRA_LOG_ERROR << "compuress data error, data: "
<< std::string(source.begin(), source.end());
data.net_err = std::make_error_code(std::errc::protocol_error);
data.status = 404;
}
}
}
else {
while (true) {
auto result = co_await source();

std::span<char> msg(result.buf.data(), result.buf.size());
std::string encode_header = ws.encode_frame(msg, op, result.eof);
else {
#endif
std::string encode_header = ws.encode_frame(source, op, true);
std::vector<asio::const_buffer> buffers{
asio::buffer(encode_header.data(), encode_header.size()),
asio::buffer(msg.data(), msg.size())};
asio::buffer(source.data(), source.size())};

auto [ec, _] = co_await async_write(buffers);
if (ec) {
data.net_err = ec;
data.status = 404;
break;
}
#ifdef CINATRA_ENABLE_GZIP
}
#endif
}
else {
while (true) {
auto result = co_await source();
#ifdef CINATRA_ENABLE_GZIP
if (enable_ws_deflate_ && is_server_support_ws_deflate_) {
std::string dest_buf;
if (cinatra::gzip_codec::deflate(
{result.buf.data(), result.buf.size()}, dest_buf)) {
std::span<char> msg(dest_buf.data(), dest_buf.size());
std::string header = ws.encode_frame(msg, op, result.eof, true);
std::vector<asio::const_buffer> buffers{asio::buffer(header),
asio::buffer(dest_buf)};
auto [ec, sz] = co_await async_write(buffers);
if (ec) {
data.net_err = ec;
data.status = 404;
}
}
else {
CINATRA_LOG_ERROR << "compuress data error, data: "
<< std::string(result.buf.data());
data.net_err = std::make_error_code(std::errc::protocol_error);
data.status = 404;
}
}
else {
#endif
std::span<char> msg(result.buf.data(), result.buf.size());
std::string encode_header = ws.encode_frame(msg, op, result.eof);
std::vector<asio::const_buffer> buffers{
asio::buffer(encode_header.data(), encode_header.size()),
asio::buffer(msg.data(), msg.size())};

if (result.eof) {
break;
auto [ec, _] = co_await async_write(buffers);
if (ec) {
data.net_err = ec;
data.status = 404;
break;
}

if (result.eof) {
break;
}
#ifdef CINATRA_ENABLE_GZIP
}
#endif
}
}

Expand Down Expand Up @@ -1839,9 +1922,28 @@ class coro_http_client : public std::enable_shared_from_this<coro_http_client> {
}
}

data.status = 200;
data.resp_body = {data_ptr, payload_len};
#ifdef CINATRA_ENABLE_GZIP
if (!is_close_frame && is_server_support_ws_deflate_ &&
enable_ws_deflate_) {
inflate_str_.clear();
if (!cinatra::gzip_codec::inflate({data_ptr, payload_len},
inflate_str_)) {
CINATRA_LOG_ERROR << "uncompuress data error";
data.status = 404;
data.net_err = std::make_error_code(std::errc::protocol_error);
co_return data;
}
data.status = 200;
data.resp_body = {inflate_str_.data(), inflate_str_.size()};
}
else {
#endif

data.status = 200;
data.resp_body = {data_ptr, payload_len};
#ifdef CINATRA_ENABLE_GZIP
}
#endif
read_buf.consume(read_buf.size());
header_size = 2;

Expand Down Expand Up @@ -2024,6 +2126,12 @@ class coro_http_client : public std::enable_shared_from_this<coro_http_client> {
std::string resp_chunk_str_;
std::span<char> out_buf_;

#ifdef CINATRA_ENABLE_GZIP
bool enable_ws_deflate_ = false;
bool is_server_support_ws_deflate_ = false;
std::string inflate_str_;
#endif

#ifdef BENCHMARK_TEST
std::string req_str_;
bool stop_bench_ = false;
Expand Down
68 changes: 63 additions & 5 deletions include/cinatra/coro_http_connection.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,9 @@
#include "sha1.hpp"
#include "string_resize.hpp"
#include "websocket.hpp"
#ifdef CINATRA_ENABLE_GZIP
#include "gzip.hpp"
#endif
#include "ylt/coro_io/coro_file.hpp"
#include "ylt/coro_io/coro_io.hpp"

Expand Down Expand Up @@ -132,6 +135,14 @@ class coro_http_connection
if (body_len == 0) {
if (parser_.method() == "GET"sv) {
if (request_.is_upgrade()) {
#ifdef CINATRA_ENABLE_GZIP
if (request_.is_support_compressed()) {
is_client_ws_compressed_ = true;
}
else {
is_client_ws_compressed_ = false;
}
#endif
// websocket
build_ws_handshake_head();
bool ok = co_await reply(true); // response ws handshake
Expand Down Expand Up @@ -562,11 +573,28 @@ class coro_http_connection

async_simple::coro::Lazy<std::error_code> write_websocket(
std::string_view msg, opcode op = opcode::text) {
auto header = ws_.format_header(msg.length(), op);
std::vector<asio::const_buffer> buffers;
buffers.push_back(asio::buffer(header));
buffers.push_back(asio::buffer(msg));
std::string header;
#ifdef CINATRA_ENABLE_GZIP
std::string dest_buf;
if (is_client_ws_compressed_ && msg.size() > 0) {
if (!cinatra::gzip_codec::deflate(msg, dest_buf)) {
CINATRA_LOG_ERROR << "compuress data error, data: " << msg;
co_return std::make_error_code(std::errc::protocol_error);
}

header = ws_.format_header(dest_buf.length(), op, true);
buffers.push_back(asio::buffer(header));
buffers.push_back(asio::buffer(dest_buf));
}
else {
#endif
header = ws_.format_header(msg.length(), op);
buffers.push_back(asio::buffer(header));
buffers.push_back(asio::buffer(msg));
#ifdef CINATRA_ENABLE_GZIP
}
#endif
auto [ec, sz] = co_await async_write(buffers);
co_return ec;
}
Expand Down Expand Up @@ -623,8 +651,27 @@ class coro_http_connection
break;
case cinatra::ws_frame_type::WS_TEXT_FRAME:
case cinatra::ws_frame_type::WS_BINARY_FRAME: {
result.eof = true;
result.data = {payload.data(), payload.size()};
#ifdef CINATRA_ENABLE_GZIP
if (is_client_ws_compressed_) {
inflate_str_.clear();
if (!cinatra::gzip_codec::inflate(
{payload.data(), payload.size()}, inflate_str_)) {
CINATRA_LOG_ERROR << "uncompuress data error";
result.ec = std::make_error_code(std::errc::protocol_error);
break;
}
result.eof = true;
result.data = {inflate_str_.data(), inflate_str_.size()};
break;
}
else {
#endif
result.eof = true;
result.data = {payload.data(), payload.size()};
break;
#ifdef CINATRA_ENABLE_GZIP
}
#endif
} break;
case cinatra::ws_frame_type::WS_CLOSE_FRAME: {
close_frame close_frame =
Expand Down Expand Up @@ -811,6 +858,12 @@ class coro_http_connection
response_.add_header("Connection", "Upgrade");
response_.add_header("Sec-WebSocket-Accept", std::string(accept_key, 28));
auto protocal_str = request_.get_header_value("sec-websocket-protocol");
#ifdef CINATRA_ENABLE_GZIP
if (is_client_ws_compressed_) {
response_.add_header("Sec-WebSocket-Extensions",
"permessage-deflate; client_no_context_takeover");
}
#endif
if (!protocal_str.empty()) {
response_.add_header("Sec-WebSocket-Protocol", std::string(protocal_str));
}
Expand All @@ -837,6 +890,11 @@ class coro_http_connection
uint64_t max_part_size_ = 8 * 1024 * 1024;
std::string resp_str_;

#ifdef CINATRA_ENABLE_GZIP
bool is_client_ws_compressed_ = false;
std::string inflate_str_;
#endif

websocket ws_;
#ifdef CINATRA_ENABLE_SSL
std::unique_ptr<asio::ssl::context> ssl_ctx_ = nullptr;
Expand Down
8 changes: 8 additions & 0 deletions include/cinatra/coro_http_request.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -208,6 +208,14 @@ class coro_http_request {
return true;
}

bool is_support_compressed() {
auto extension_str = get_header_value("Sec-WebSocket-Extensions");
if (extension_str.find("permessage-deflate") != std::string::npos) {
return true;
}
return false;
}

void set_aspect_data(std::string data) {
aspect_data_.push_back(std::move(data));
}
Expand Down
Loading
Loading