From cfe3e286b3c1c8ac62b6ec75b1544126fb8cfaeb Mon Sep 17 00:00:00 2001 From: helintong Date: Thu, 18 Apr 2024 11:22:12 +0800 Subject: [PATCH] feat: add websocket permessage-deflate --- include/cinatra/coro_http_client.hpp | 154 ++++++++++++++++++---- include/cinatra/coro_http_connection.hpp | 65 +++++++++ include/cinatra/coro_http_request.hpp | 8 ++ include/cinatra/gzip.hpp | 159 +++++++++++++++++++++++ include/cinatra/websocket.hpp | 17 ++- tests/test_cinatra_websocket.cpp | 70 ++++++++++ 6 files changed, 444 insertions(+), 29 deletions(-) diff --git a/include/cinatra/coro_http_client.hpp b/include/cinatra/coro_http_client.hpp index 720b21e3..e68dc1f9 100644 --- a/include/cinatra/coro_http_client.hpp +++ b/include/cinatra/coro_http_client.hpp @@ -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" @@ -274,7 +277,7 @@ class coro_http_client : public std::enable_shared_from_this { } // only make socket connet(or handshake) to the host - async_simple::coro::Lazy connect(std::string uri) { + async_simple::coro::Lazy connect(std::string uri, bool enable_ws_deflate = false) { resp_data data{}; bool no_schema = !has_schema(uri); std::string append_uri; @@ -298,10 +301,30 @@ class coro_http_client : public std::enable_shared_from_this { } 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 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); @@ -370,39 +393,98 @@ class coro_http_client : public std::enable_shared_from_this { } if constexpr (is_span_v) { - std::string encode_header = ws.encode_frame(source, op, need_mask); - std::vector 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( + std::string(source.begin(), source.end()), dest_buf)) { + std::span msg(dest_buf.data(), dest_buf.size()); + auto header = ws.encode_frame(msg, op, need_mask, true, true); + std::vector 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 { - while (true) { - auto result = co_await source(); - - std::span msg(result.buf.data(), result.buf.size()); - std::string encode_header = - ws.encode_frame(msg, op, need_mask, result.eof); + else { +#endif + std::string encode_header = ws.encode_frame(source, op, need_mask); std::vector 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(); - if (result.eof) { - break; +#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 msg(dest_buf.data(), dest_buf.size()); + std::string header = + ws.encode_frame(msg, op, need_mask, result.eof, true); + std::vector 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 msg(result.buf.data(), result.buf.size()); + std::string encode_header = + ws.encode_frame(msg, op, need_mask, result.eof); + std::vector buffers{ + asio::buffer(encode_header.data(), encode_header.size()), + asio::buffer(msg.data(), msg.size())}; + + 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 } co_return data; @@ -1828,8 +1910,27 @@ class coro_http_client : public std::enable_shared_from_this { } } - 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_) { + 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; @@ -2013,6 +2114,11 @@ class coro_http_client : public std::enable_shared_from_this { std::string resp_chunk_str_; std::span 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; diff --git a/include/cinatra/coro_http_connection.hpp b/include/cinatra/coro_http_connection.hpp index 6f25a9f1..0a78f0a8 100644 --- a/include/cinatra/coro_http_connection.hpp +++ b/include/cinatra/coro_http_connection.hpp @@ -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" @@ -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 @@ -551,6 +562,32 @@ class coro_http_connection async_simple::coro::Lazy 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 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 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 buffers; buffers.push_back(asio::buffer(header)); @@ -558,6 +595,7 @@ class coro_http_connection auto [ec, sz] = co_await async_write(buffers); co_return ec; +#endif } async_simple::coro::Lazy read_websocket() { @@ -612,8 +650,26 @@ 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 = @@ -803,6 +859,12 @@ 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: @@ -825,6 +887,9 @@ class coro_http_connection std::atomic 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 diff --git a/include/cinatra/coro_http_request.hpp b/include/cinatra/coro_http_request.hpp index 36309a12..ea6574fd 100644 --- a/include/cinatra/coro_http_request.hpp +++ b/include/cinatra/coro_http_request.hpp @@ -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)); } diff --git a/include/cinatra/gzip.hpp b/include/cinatra/gzip.hpp index 400ce6ff..23a9e78c 100644 --- a/include/cinatra/gzip.hpp +++ b/include/cinatra/gzip.hpp @@ -140,4 +140,163 @@ inline int uncompress_file(const char *src_file, const char *out_file_name) { return 0; } + +bool inflate(const std::string &str_src, std::string &str_dest) { + int err = Z_DATA_ERROR; + // Create stream + z_stream zs = {0}; + // Set output data streams, do this here to avoid overwriting on recursive + // calls + const int OUTPUT_BUF_SIZE = 8192; + Bytef bytes_out[OUTPUT_BUF_SIZE] = {0}; + + // Initialise the z_stream + err = ::inflateInit2(&zs, -15); + if (err != Z_OK) { + return false; + } + + // Use whatever input is provided + zs.next_in = (Bytef *)(str_src.c_str()); + zs.avail_in = str_src.length(); + + do { + try { + // Initialise stream values + // zs->zalloc = (alloc_func)0; + // zs->zfree = (free_func)0; + // zs->opaque = (voidpf)0; + + zs.next_out = bytes_out; + zs.avail_out = OUTPUT_BUF_SIZE; + + // Try to unzip the data + err = ::inflate(&zs, Z_SYNC_FLUSH); + + // Is zip finished reading all currently available input and writing all + // generated output + if (err == Z_STREAM_END) { + // Finish up + int kerr = ::inflateEnd(&zs); + + // Got a good result, set the size to the amount unzipped in this call + // (including all recursive calls) + + str_dest.append((const char *)bytes_out, + OUTPUT_BUF_SIZE - zs.avail_out); + return true; + } + else if ((err == Z_OK) && (zs.avail_out == 0) && (zs.avail_in != 0)) { + // Output array was not big enough, call recursively until there is + // enough space + + str_dest.append((const char *)bytes_out, + OUTPUT_BUF_SIZE - zs.avail_out); + + continue; + } + else if ((err == Z_OK) && (zs.avail_in == 0)) { + // All available input has been processed, everything ok. + // Set the size to the amount unzipped in this call (including all + // recursive calls) + str_dest.append((const char *)bytes_out, + OUTPUT_BUF_SIZE - zs.avail_out); + + int kerr = ::inflateEnd(&zs); + + break; + } + else { + return false; + } + } catch (...) { + return false; + } + } while (true); + + return err == Z_OK; +} + +bool deflate(const std::string &str_src, std::string &str_dest) { + int err = Z_DATA_ERROR; + // Create stream + z_stream zs = {0}; + // Set output data streams, do this here to avoid overwriting on recursive + // calls + const int OUTPUT_BUF_SIZE = 8192; + Bytef bytes_out[OUTPUT_BUF_SIZE] = {0}; + + // Initialise the z_stream + err = ::deflateInit2(&zs, 1, Z_DEFLATED, -15, 8, Z_DEFAULT_STRATEGY); + if (err != Z_OK) { + return false; + } + // Use whatever input is provided + zs.next_in = (Bytef *)(str_src.c_str()); + zs.avail_in = str_src.length(); + + do { + try { + // Initialise stream values + // zs->zalloc = (alloc_func)0; + // zs->zfree = (free_func)0; + // zs->opaque = (voidpf)0; + + zs.next_out = bytes_out; + zs.avail_out = OUTPUT_BUF_SIZE; + + // Try to unzip the data + err = ::deflate(&zs, Z_SYNC_FLUSH); + + // Is zip finished reading all currently available input and writing all + // generated output + if (err == Z_STREAM_END) { + // Finish up + int kerr = ::deflateEnd(&zs); + + // Got a good result, set the size to the amount unzipped in this call + // (including all recursive calls) + + str_dest.append((const char *)bytes_out, + OUTPUT_BUF_SIZE - zs.avail_out); + return true; + } + else if ((err == Z_OK) && (zs.avail_out == 0) && (zs.avail_in != 0)) { + // Output array was not big enough, call recursively until there is + // enough space + + str_dest.append((const char *)bytes_out, + OUTPUT_BUF_SIZE - zs.avail_out); + + continue; + } + else if ((err == Z_OK) && (zs.avail_in == 0)) { + // All available input has been processed, everything ok. + // Set the size to the amount unzipped in this call (including all + // recursive calls) + str_dest.append((const char *)bytes_out, + OUTPUT_BUF_SIZE - zs.avail_out); + + int kerr = ::deflateEnd(&zs); + + break; + } + else { + return false; + } + } catch (...) { + return false; + } + } while (true); + + if (err == Z_OK) { + // subtract 4 to remove the extra 00 00 ff ff added to the end of the deflat + // function + str_dest = str_dest.substr(0, str_dest.length() - 4); + return true; + } + + return false; +} + } // namespace cinatra::gzip_codec \ No newline at end of file diff --git a/include/cinatra/websocket.hpp b/include/cinatra/websocket.hpp index 8327ef2e..1b7ee2d2 100644 --- a/include/cinatra/websocket.hpp +++ b/include/cinatra/websocket.hpp @@ -121,19 +121,23 @@ class websocket { return ws_frame_type::WS_BINARY_FRAME; } - std::string format_header(size_t length, opcode code) { - size_t header_length = encode_header(length, code); + std::string format_header(size_t length, opcode code, + bool is_compressed = false) { + size_t header_length = encode_header(length, code, is_compressed); return {msg_header_, header_length}; } std::string encode_frame(std::span &data, opcode op, bool need_mask, - bool eof = true) { + bool eof = true, bool need_compression = false) { std::string header; /// Base header. frame_header hdr{}; hdr.fin = eof; hdr.rsv1 = 0; - hdr.rsv2 = 0; + if (need_compression) + hdr.rsv2 = 1; + else + hdr.rsv2 = 0; hdr.rsv3 = 0; hdr.opcode = static_cast(op); hdr.mask = 1; @@ -227,7 +231,7 @@ class websocket { opcode get_opcode() { return (opcode)msg_opcode_; } private: - size_t encode_header(size_t length, opcode code) { + size_t encode_header(size_t length, opcode code, bool is_compressed = false) { size_t header_length; if (length < 126) { @@ -251,6 +255,9 @@ class websocket { msg_header_[0] |= code; } + if (is_compressed) + msg_header_[0] |= 0x40; + return header_length; } diff --git a/tests/test_cinatra_websocket.cpp b/tests/test_cinatra_websocket.cpp index 124c0816..26b58169 100644 --- a/tests/test_cinatra_websocket.cpp +++ b/tests/test_cinatra_websocket.cpp @@ -278,3 +278,73 @@ TEST_CASE("test client quit after send msg") { async_simple::coro::syncAwait(test_websocket()); } + +#ifdef CINATRA_ENABLE_GZIP +TEST_CASE("test websocket permessage defalte") { + coro_http_server server(1, 8090); + server.set_http_handler( + "/ws_extesion", + [](coro_http_request &req, + coro_http_response &resp) -> async_simple::coro::Lazy { + websocket_result result{}; + while (true) { + result = co_await req.get_conn()->read_websocket(); + if (result.ec) { + break; + } + + if (result.type == ws_frame_type::WS_CLOSE_FRAME) { + std::cout << "close frame\n"; + break; + } + + if (result.type == ws_frame_type::WS_TEXT_FRAME || + result.type == ws_frame_type::WS_BINARY_FRAME) { + CHECK(result.data == "test"); + } + else if (result.type == ws_frame_type::WS_PING_FRAME || + result.type == ws_frame_type::WS_PONG_FRAME) { + // ping pong frame just need to continue, no need echo anything, + // because framework has reply ping/pong msg to client + // automatically. + continue; + } + else { + // error frame + break; + } + + auto ec = co_await req.get_conn()->write_websocket(result.data); + if (ec) { + break; + } + } + }); + + server.async_start(); + + coro_http_client client{}; + REQUIRE(async_simple::coro::syncAwait( + client.connect("ws://localhost:8090/ws_extesion", true))); + + std::string send_str("test"); + + client.on_ws_msg([&, send_str](resp_data data) { + if (data.net_err) { + std::cout << "ws_msg net error " << data.net_err.message() << "\n"; + return; + } + + std::cout << "ws msg len: " << data.resp_body.size() << std::endl; + REQUIRE(data.resp_body.size() == send_str.size()); + CHECK(data.resp_body == send_str); + }); + + async_simple::coro::syncAwait(client.write_websocket(send_str)); + + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + + server.stop(); + client.close(); +} +#endif \ No newline at end of file