Skip to content

Commit

Permalink
feat: add permessage deflate extension for websocket
Browse files Browse the repository at this point in the history
  • Loading branch information
@helintongh
helintongh committed Apr 17, 2024
1 parent e1e204f commit ae21eec
Show file tree
Hide file tree
Showing 6 changed files with 424 additions and 6 deletions.
109 changes: 108 additions & 1 deletion include/cinatra/coro_http_client.hpp
View file
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 @@ -319,14 +322,16 @@ class coro_http_client : public std::enable_shared_from_this<coro_http_client> {

void set_ws_sec_key(std::string sec_key) { ws_sec_key_ = std::move(sec_key); }

async_simple::coro::Lazy<bool> async_ws_connect(std::string uri) {
async_simple::coro::Lazy<bool> async_ws_connect(std::string uri, bool enable_ws_deflate = false) {
resp_data data{};
auto [r, u] = handle_uri(data, uri);
if (!r) {
CINATRA_LOG_WARNING << "url error:";
co_return false;
}

enable_ws_deflate_ = enable_ws_deflate;

req_context<> ctx{};
if (u.is_websocket()) {
// build websocket http header
Expand All @@ -337,10 +342,33 @@ 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
add_header("Sec-WebSocket-Extensions", "permessage-deflate; client_max_window_bits");
#endif
}

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) {
std::cout << c.name << " value is: " << c.value << std::endl;
if (c.name == "Sec-WebSocket-Extensions") {
std::cout << "have extensions\n";
if (c.value.find("permessage-deflate;") != std::string::npos) {
std::cout << "support deflate extensions\n";
is_server_support_ws_deflate_ = true;
}
else {
std::cout << "not support deflate extensions\n";
is_server_support_ws_deflate_ = false;
}
break;
}
}
}
#endif

async_read_ws().start([](auto &&) {
});
co_return !data.net_err;
Expand Down Expand Up @@ -376,6 +404,30 @@ class coro_http_client : public std::enable_shared_from_this<coro_http_client> {
}

if constexpr (is_span_v<Source>) {
#ifdef CINATRA_ENABLE_GZIP
if (enable_ws_deflate_ && is_server_support_ws_deflate_) {
std::string dest_buf;
if (cinatra::gzip_codec::deflate(std::string(source.begin(), source.end()), dest_buf)) {
std::span<char> msg(dest_buf.data(), dest_buf.size());
auto header = ws.encode_frame(msg, op, need_mask, true, true);
std::vector<asio::const_buffer> buffers;
buffers.push_back(asio::buffer(header));
buffers.push_back(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 {
#endif
std::string encode_header = ws.encode_frame(source, op, need_mask);
std::vector<asio::const_buffer> buffers{
asio::buffer(encode_header.data(), encode_header.size()),
Expand All @@ -386,11 +438,40 @@ class coro_http_client : public std::enable_shared_from_this<coro_http_client> {
data.net_err = ec;
data.status = 404;
}
#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(std::string(result), dest_buf)) {
std::span<char> msg(dest_buf.data(), dest_buf.size());
std::string header =
ws.encode_frame(msg, op, need_mask, result.eof, true);
std::vector<asio::const_buffer> buffers;
buffers.push_back(asio::buffer(header));
buffers.push_back(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 {
#endif

std::span<char> msg(result.buf.data(), result.buf.size());
std::string encode_header =
ws.encode_frame(msg, op, need_mask, result.eof);
Expand All @@ -409,6 +490,9 @@ class coro_http_client : public std::enable_shared_from_this<coro_http_client> {
break;
}
}
#ifdef CINATRA_ENABLE_GZIP
}
#endif
}

co_return data;
Expand Down Expand Up @@ -1848,9 +1932,27 @@ class coro_http_client : public std::enable_shared_from_this<coro_http_client> {
data_ptr += sizeof(uint16_t);
}
}
#ifdef CINATRA_ENABLE_GZIP
if (!is_close_frame && is_server_support_ws_deflate_ && enable_ws_deflate_) {
std::string out;
if (!cinatra::gzip_codec::inflate(std::string(data_ptr), out))
{
CINATRA_LOG_ERROR << "uncompuress data error";
data.status = 404;
data.net_err = std::make_error_code(std::errc::protocol_error);
break;
}

data.status = 200;
data.resp_body = {out.data(), out.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 @@ -2042,6 +2144,11 @@ class coro_http_client : public std::enable_shared_from_this<coro_http_client> {
std::string resp_chunk_str_;
std::span<char> out_buf_;

bool enable_ws_deflate_ = false;
#ifdef CINATRA_ENABLE_GZIP
bool is_server_support_ws_deflate_ = false;
#endif

#ifdef BENCHMARK_TEST
std::string req_str_;
bool stop_bench_ = false;
Expand Down
65 changes: 65 additions & 0 deletions include/cinatra/coro_http_connection.hpp
View file
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 @@ -551,13 +562,40 @@ class coro_http_connection

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

auto header = ws_.format_header(dest_buf.length(), op, true);
std::vector<asio::const_buffer> buffers;
buffers.push_back(asio::buffer(header));
buffers.push_back(asio::buffer(dest_buf));

auto [ec, sz] = co_await async_write(buffers);
co_return ec;
}
else {
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));

auto [ec, sz] = co_await async_write(buffers);
co_return ec;
}
#else
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));

auto [ec, sz] = co_await async_write(buffers);
co_return ec;
#endif
}

async_simple::coro::Lazy<websocket_result> read_websocket() {
Expand Down Expand Up @@ -612,8 +650,27 @@ class coro_http_connection
break;
case cinatra::ws_frame_type::WS_TEXT_FRAME:
case cinatra::ws_frame_type::WS_BINARY_FRAME: {
#ifdef CINATRA_ENABLE_GZIP
std::string out;
if (is_client_ws_compressed_)
{
if (!cinatra::gzip_codec::inflate(std::string(payload.begin(), payload.end()), out))
{
CINATRA_LOG_ERROR << "uncompuress data error";
result.ec = std::make_error_code(std::errc::protocol_error);
break;
}
result.eof = true;
result.data = {out.data(), out.size()};
}
else {
result.eof = true;
result.data = {payload.data(), payload.size()};
}
#else
result.eof = true;
result.data = {payload.data(), payload.size()};
#endif
} break;
case cinatra::ws_frame_type::WS_CLOSE_FRAME: {
close_frame close_frame =
Expand Down Expand Up @@ -803,6 +860,11 @@ class coro_http_connection
if (!protocal_str.empty()) {
response_.add_header("Sec-WebSocket-Protocol", std::string(protocal_str));
}
#ifdef CINATRA_ENABLE_GZIP
if (is_client_ws_compressed_) {
response_.add_header("Sec-WebSocket-Extensions", "permessage-deflate; client_no_context_takeover");
}
#endif
}

private:
Expand All @@ -825,6 +887,9 @@ class coro_http_connection
std::atomic<std::chrono::system_clock::time_point> last_rwtime_;
uint64_t max_part_size_ = 8 * 1024 * 1024;
std::string resp_str_;
#ifdef CINATRA_ENABLE_GZIP
bool is_client_ws_compressed_ = false;
#endif

websocket ws_;
#ifdef CINATRA_ENABLE_SSL
Expand Down
8 changes: 8 additions & 0 deletions include/cinatra/coro_http_request.hpp
View file
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

0 comments on commit ae21eec

Please sign in to comment.