Skip to content

Commit 60f421a

Browse files
lupin012lupin012
authored
rpcdaemon: add ws compression (#1798)
1 parent ed1b934 commit 60f421a

File tree

9 files changed

+47
-15
lines changed

9 files changed

+47
-15
lines changed

‎cmd/common/rpcdaemon_options.cpp‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -108,9 +108,13 @@ void add_rpcdaemon_options(CLI::App& cli, silkworm::rpc::DaemonSettings& setting
108108
->description("Flag indicating if strict compatibility with Erigon RpcDaemon is enabled")
109109
->capture_default_str();
110110

111-
cli.add_flag("--websocket", settings.use_websocket)
111+
cli.add_flag("--ws", settings.use_websocket)
112112
->description("Enable WebSocket protocol for Execution Layer and Engine JSON RPC API, same port as HTTP(S)")
113113
->capture_default_str();
114+
115+
cli.add_flag("--ws-compression", settings.ws_compression)
116+
->description("Enable compression on WebSocket protocol for Execution Layer and Engine JSON RPC API")
117+
->capture_default_str();
114118
}
115119

116120
} // namespace silkworm::cmd::common

‎silkworm/rpc/daemon.cpp‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -302,12 +302,14 @@ void Daemon::start() {
302302
if (not settings_.eth_end_point.empty()) {
303303
rpc_services_.emplace_back(
304304
std::make_unique<http::Server>(
305-
settings_.eth_end_point, settings_.eth_api_spec, ioc, worker_pool_, settings_.cors_domain, /*jwt_secret=*/std::nullopt, settings_.use_websocket));
305+
settings_.eth_end_point, settings_.eth_api_spec, ioc, worker_pool_, settings_.cors_domain, /*jwt_secret=*/std::nullopt,
306+
settings_.use_websocket, settings_.ws_compression));
306307
}
307308
if (not settings_.engine_end_point.empty()) {
308309
rpc_services_.emplace_back(
309310
std::make_unique<http::Server>(
310-
settings_.engine_end_point, kDefaultEth2ApiSpec, ioc, worker_pool_, settings_.cors_domain, jwt_secret_, settings_.use_websocket));
311+
settings_.engine_end_point, kDefaultEth2ApiSpec, ioc, worker_pool_, settings_.cors_domain, jwt_secret_,
312+
settings_.use_websocket, settings_.ws_compression));
311313
}
312314
}
313315

‎silkworm/rpc/http/connection.cpp‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -38,14 +38,16 @@ Connection::Connection(boost::asio::io_context& io_context,
3838
commands::RpcApiTable& handler_table,
3939
const std::vector<std::string>& allowed_origins,
4040
std::optional<std::string> jwt_secret,
41-
bool use_websocket)
41+
bool use_websocket,
42+
bool ws_compression)
4243
: socket_{io_context},
4344
api_{api},
4445
handler_table_{handler_table},
4546
request_handler_{this, api, handler_table},
4647
allowed_origins_{allowed_origins},
4748
jwt_secret_{std ::move(jwt_secret)},
48-
use_websocket_{use_websocket} {
49+
use_websocket_{use_websocket},
50+
ws_compression_{ws_compression} {
4951
SILK_TRACE << "Connection::Connection socket " << &socket_ << " created";
5052
}
5153

@@ -100,7 +102,7 @@ Task<void> Connection::do_upgrade(const boost::beast::http::request<boost::beast
100102
// we tie the socket object to a websocket stream
101103
boost::beast::websocket::stream<boost::beast::tcp_stream> stream(std::move(socket_));
102104

103-
auto ws_connection = std::make_shared<ws::Connection>(std::move(stream), api_, std::move(handler_table_));
105+
auto ws_connection = std::make_shared<ws::Connection>(std::move(stream), api_, std::move(handler_table_), ws_compression_);
104106
co_await ws_connection->accept(req);
105107

106108
auto connection_loop = [](auto websocket_connection) -> Task<void> { co_await websocket_connection->read_loop(); };

‎silkworm/rpc/http/connection.hpp‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,8 @@ class Connection : public Channel {
4848
commands::RpcApiTable& handler_table,
4949
const std::vector<std::string>& allowed_origins,
5050
std::optional<std::string> jwt_secret,
51-
bool use_websocket);
51+
bool use_websocket,
52+
bool ws_compression);
5253
~Connection() override;
5354

5455
boost::asio::ip::tcp::socket& socket() { return socket_; }
@@ -100,6 +101,8 @@ class Connection : public Channel {
100101
boost::beast::flat_buffer data_;
101102

102103
bool use_websocket_;
104+
105+
bool ws_compression_;
103106
};
104107

105108
} // namespace silkworm::rpc::http

‎silkworm/rpc/http/server.cpp‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -47,14 +47,16 @@ Server::Server(const std::string& end_point,
4747
boost::asio::thread_pool& workers,
4848
std::vector<std::string> allowed_origins,
4949
std::optional<std::string> jwt_secret,
50-
bool use_websocket)
50+
bool use_websocket,
51+
bool ws_compression)
5152
: rpc_api_{io_context, workers},
5253
handler_table_{api_spec},
5354
io_context_(io_context),
5455
acceptor_{io_context},
5556
allowed_origins_{std::move(allowed_origins)},
5657
jwt_secret_(std::move(jwt_secret)),
57-
use_websocket_{use_websocket} {
58+
use_websocket_{use_websocket},
59+
ws_compression_{ws_compression} {
5860
const auto [host, port] = parse_endpoint(end_point);
5961

6062
// Open the acceptor with the option to reuse the address (i.e. SO_REUSEADDR).
@@ -79,7 +81,8 @@ Task<void> Server::run() {
7981
while (acceptor_.is_open()) {
8082
SILK_DEBUG << "Server::run accepting using io_context " << &io_context_ << "...";
8183

82-
auto new_connection = std::make_shared<Connection>(io_context_, rpc_api_, handler_table_, allowed_origins_, jwt_secret_, use_websocket_);
84+
auto new_connection = std::make_shared<Connection>(io_context_, rpc_api_, handler_table_, allowed_origins_, jwt_secret_,
85+
use_websocket_, ws_compression_);
8386
co_await acceptor_.async_accept(new_connection->socket(), boost::asio::use_awaitable);
8487
if (!acceptor_.is_open()) {
8588
SILK_TRACE << "Server::run returning...";

‎silkworm/rpc/http/server.hpp‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -51,7 +51,8 @@ class Server {
5151
boost::asio::thread_pool& workers,
5252
std::vector<std::string> allowed_origins,
5353
std::optional<std::string> jwt_secret,
54-
bool use_websocket);
54+
bool use_websocket,
55+
bool compression);
5556

5657
void start();
5758

@@ -81,6 +82,8 @@ class Server {
8182
std::optional<std::string> jwt_secret_;
8283

8384
bool use_websocket_;
85+
86+
bool ws_compression_;
8487
};
8588

8689
} // namespace silkworm::rpc::http

‎silkworm/rpc/settings.hpp‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@ struct DaemonSettings {
4040
bool skip_protocol_check{false};
4141
bool erigon_json_rpc_compatibility{false};
4242
bool use_websocket{false};
43+
bool ws_compression{false};
4344
};
4445

4546
} // namespace silkworm::rpc

‎silkworm/rpc/ws/connection.cpp‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -30,9 +30,11 @@ namespace silkworm::rpc::ws {
3030

3131
Connection::Connection(boost::beast::websocket::stream<boost::beast::tcp_stream>&& stream,
3232
commands::RpcApi& api,
33-
const commands::RpcApiTable& handler_table)
33+
const commands::RpcApiTable& handler_table,
34+
bool compression)
3435
: ws_{std::move(stream)},
35-
request_handler_{this, api, handler_table} {
36+
request_handler_{this, api, handler_table},
37+
compression_{compression} {
3638
SILK_DEBUG << "ws::Connection::Connection ws created:" << &ws_;
3739
}
3840

@@ -43,6 +45,13 @@ Connection::~Connection() {
4345
Task<void> Connection::accept(const boost::beast::http::request<boost::beast::http::string_body>& req) {
4446
// Set suggested timeout settings for the websocket
4547
ws_.set_option(boost::beast::websocket::stream_base::timeout::suggested(boost::beast::role_type::server));
48+
if (compression_) {
49+
boost::beast::websocket::permessage_deflate opt{
50+
.server_enable = true,
51+
.client_enable = true,
52+
};
53+
ws_.set_option(opt);
54+
}
4655

4756
// Accept the websocket handshake
4857
co_await ws_.async_accept(req, boost::asio::use_awaitable);

‎silkworm/rpc/ws/connection.hpp‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -44,17 +44,18 @@ class Connection : public Channel {
4444
//! Construct a connection running within the given execution context.
4545
Connection(boost::beast::websocket::stream<boost::beast::tcp_stream>&& stream,
4646
commands::RpcApi& api,
47-
const commands::RpcApiTable& handler_table);
47+
const commands::RpcApiTable& handler_table,
48+
bool compression = false);
4849

4950
~Connection() override;
5051

5152
Task<void> accept(const boost::beast::http::request<boost::beast::http::string_body>& req);
5253

5354
Task<void> read_loop();
5455

56+
// Methods of Channel interface
5557
Task<void> open_stream() override { co_return; }
5658
Task<void> close() override { co_return; }
57-
5859
Task<void> write_rsp(const std::string& content) override;
5960
Task<std::size_t> write(std::string_view content) override;
6061

@@ -64,10 +65,14 @@ class Connection : public Channel {
6465
//! Perform an asynchronous write operation.
6566
Task<std::size_t> do_write(const std::string& content);
6667

68+
// websocket stream
6769
boost::beast::websocket::stream<boost::beast::tcp_stream> ws_;
6870

6971
//! The handler used to process the incoming request.
7072
http::RequestHandler request_handler_;
73+
74+
//! enable compress flag
75+
bool compression_{false};
7176
};
7277

7378
} // namespace silkworm::rpc::ws

0 commit comments

Comments
 (0)