Skip to content

Commit 0e4125d

Browse files
TzuHuanTaiTzuHuanTai
committed
refactor: use variant
1 parent 57db5bc commit 0e4125d

File tree

9 files changed

+225
-135
lines changed

9 files changed

+225
-135
lines changed

‎.vscode/c_cpp_properties.json‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@
1111
"/usr/local/include/webrtc/third_party/abseil-cpp/*",
1212
"/usr/local/include/webrtc/third_party/libyuv/include/*",
1313
"/usr/local/include/webrtc/tools/json_schema_compiler/*",
14+
"/usr/local/include/webrtc/third_party/boringssl/src/include/*",
1415
"/usr/include/boost/program_options/*",
1516
"${workspaceFolder}/src"
1617
],

‎CMakeLists.txt‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@ add_compile_definitions(
3636
WEBRTC_LINUX
3737
WEBRTC_USE_H264
3838
USE_CPPRESTSDK
39+
OPENSSL_IS_BORINGSSL
3940
)
4041

4142
add_compile_options(
@@ -48,6 +49,7 @@ include_directories(
4849
${WEBRTC_INCLUDE_DIR}
4950
${WEBRTC_INCLUDE_DIR}/third_party/abseil-cpp
5051
${WEBRTC_INCLUDE_DIR}/third_party/libyuv/include
52+
${WEBRTC_INCLUDE_DIR}/third_party/boringssl/src/include
5153
${WEBRTC_INCLUDE_DIR}/tools/json_schema_compiler
5254
)
5355

@@ -147,8 +149,6 @@ else()
147149
add_executable(${PROJECT_NAME} src/main.cpp)
148150
target_link_libraries(${PROJECT_NAME}
149151
src
150-
)
151-
target_link_libraries(${PROJECT_NAME}
152152
${WEBRTC_LINK_LIBS}
153153
boost_program_options
154154
Threads::Threads

‎src/args.h‎

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -43,14 +43,13 @@ struct Args {
4343
uint16_t http_port = 8080;
4444

4545
// websocket signaling
46-
int ws_port = 8080;
46+
bool use_tls = false;
4747
std::string ws_host = "192.168.4.21";
4848
std::string ws_token =
4949
"eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9."
50-
"eyJleHAiOjE3NzQ3NDU2MzksImlzcyI6IkFQSVduUVRzNHRtVVp2QSIsIm5iZiI6MTc0MzIwOTYzOSwic3ViIjoiZj"
51-
"JkNzRiOTUtMmYxNi00ODRiLTg3NjctYThjNWY3NzFlZWY2IiwidmlkZW8iOnsiY2FuUHVibGlzaCI6dHJ1ZSwiY2Fu"
52-
"UHVibGlzaERhdGEiOnRydWUsImNhblN1YnNjcmliZSI6ZmFsc2UsInJvb20iOiJkZXZpY2UtMSIsInJvb21Kb2luIj"
53-
"p0cnVlfX0.o7e-gjkqfMpeDjATwaWWLKUWPa8RlaoIOcuw3p8FxQk";
50+
"eyJleHAiOjE3NzM4OTI2NjEsImlzcyI6IkFQSVduUVRzNHRtVVp2QSIsIm5iZiI6MTc0MjM1NjY2MSwic3ViIj"
51+
"oiMTMxZGZjMzItNjRlMi00YjZiLTllZGEtZDdjYTU5NTNjYWJlIiwidmlkZW8iOnsicm9vbSI6ImRldmljZS0x"
52+
"Iiwicm9vbUpvaW4iOnRydWV9fQ.ThQTHYd8CBR0t3epwcak6oaleeu760V96UF8GbOMUks";
5453
};
5554

5655
#endif // ARGS_H_

‎src/main.cpp‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -25,17 +25,17 @@ int main(int argc, char *argv[]) {
2525
DEBUG_PRINT("Recorder is not started!");
2626
}
2727

28-
boost::asio::io_context ioc_;
29-
boost::asio::executor_work_guard<boost::asio::io_context::executor_type> work_guard(
30-
ioc_.get_executor());
28+
boost::asio::io_context ioc;
29+
auto work_guard = boost::asio::make_work_guard(ioc);
30+
3131
std::vector<std::shared_ptr<SignalingService>> services;
3232

3333
if (args.use_whep) {
34-
services.push_back(HttpService::Create(args, conductor, ioc_));
34+
services.push_back(HttpService::Create(args, conductor, ioc));
3535
}
3636

3737
if (args.use_websocket) {
38-
services.push_back(WebsocketService::Create(args, conductor, ioc_));
38+
services.push_back(WebsocketService::Create(args, conductor, ioc));
3939
}
4040

4141
if (args.use_mqtt) {
@@ -51,7 +51,7 @@ int main(int argc, char *argv[]) {
5151
service->Start();
5252
}
5353

54-
ioc_.run();
54+
ioc.run();
5555

5656
return 0;
5757
}

‎src/parser.cpp‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -61,7 +61,6 @@ void Parser::ParseArgs(int argc, char *argv[], Args &args) {
6161
("http_port", bpo::value<uint16_t>()->default_value(args.http_port), "Http server port")
6262
("ws_host", bpo::value<std::string>()->default_value(args.ws_host),
6363
"Websocket server host")
64-
("ws_port", bpo::value<int>()->default_value(args.ws_port), "Websocket server port")
6564
("ws_token", bpo::value<std::string>()->default_value(args.ws_token),
6665
"Websocket server token")
6766
("record_path", bpo::value<std::string>()->default_value(args.record_path),
@@ -74,6 +73,8 @@ void Parser::ParseArgs(int argc, char *argv[], Args &args) {
7473
"Use whep to exchange sdp and ice candidates")
7574
("use_websocket", bpo::bool_switch()->default_value(args.use_websocket),
7675
"Use websocket to exchange sdp and ice candidates")
76+
("use_tls", bpo::bool_switch()->default_value(args.use_tls),
77+
"Use TLS for websocket connection")
7778
("v4l2_format", bpo::value<std::string>()->default_value(args.v4l2_format),
7879
"Set v4l2 camera capture format to `i420`, `mjpeg`, `h264`. The `h264` can pass "
7980
"packets into mp4 without encoding to reduce cpu usage."
@@ -114,7 +115,6 @@ void Parser::ParseArgs(int argc, char *argv[], Args &args) {
114115
SetIfExists(vm, "mqtt_password", args.mqtt_password);
115116
SetIfExists(vm, "http_port", args.http_port);
116117
SetIfExists(vm, "ws_host", args.ws_host);
117-
SetIfExists(vm, "ws_port", args.ws_port);
118118
SetIfExists(vm, "ws_token", args.ws_token);
119119
SetIfExists(vm, "record_path", args.record_path);
120120

@@ -124,6 +124,7 @@ void Parser::ParseArgs(int argc, char *argv[], Args &args) {
124124
args.use_mqtt = vm["use_mqtt"].as<bool>();
125125
args.use_whep = vm["use_whep"].as<bool>();
126126
args.use_websocket = vm["use_websocket"].as<bool>();
127+
args.use_tls = vm["use_tls"].as<bool>();
127128

128129
if (!args.stun_url.empty() && args.stun_url.substr(0, 4) != "stun") {
129130
std::cout << "Stun url should not be empty and start with \"stun:\"" << std::endl;

‎src/signaling/CMakeLists.txt‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,4 +11,4 @@ add_library(${PROJECT_NAME}
1111
websocket_service.cpp
1212
)
1313

14-
target_link_libraries(${PROJECT_NAME} PUBLIC ${MOSQUITTO_LIBS})
14+
target_link_libraries(${PROJECT_NAME} PRIVATE ${MOSQUITTO_LIBS})

‎src/signaling/websocket_service.cpp‎

Lines changed: 135 additions & 80 deletions
Original file line numberDiff line numberDiff line change
@@ -2,111 +2,142 @@
22

33
#include <nlohmann/json.hpp>
44

5-
#include "common/logging.h"
6-
7-
const int MAX_RETRIES = 10;
8-
9-
std::shared_ptr<WebsocketService> WebsocketService::Create(Args args,
10-
std::shared_ptr<Conductor> conductor,
11-
boost::asio::io_context &ioc) {
5+
std::shared_ptr<WebsocketService>
6+
WebsocketService::Create(Args args, std::shared_ptr<Conductor> conductor, net::io_context &ioc) {
127
return std::make_shared<WebsocketService>(args, conductor, ioc);
138
}
149

1510
WebsocketService::WebsocketService(Args args, std::shared_ptr<Conductor> conductor,
16-
boost::asio::io_context &ioc)
11+
net::io_context &ioc)
1712
: SignalingService(conductor),
1813
args_(args),
19-
resolver_(ioc),
20-
ws_(ioc) {}
14+
ws_(InitWebSocket(ioc)),
15+
resolver_(net::make_strand(ioc)) {}
2116

2217
WebsocketService::~WebsocketService() { Disconnect(); }
2318

24-
void WebsocketService::OnRemoteIce(const std::string &message) {
25-
nlohmann::json res = nlohmann::json::parse(message);
26-
std::string target = res["target"];
27-
std::string canditateInit = res["candidateInit"];
19+
WebSocketVariant WebsocketService::InitWebSocket(net::io_context &ioc) {
20+
if (args_.use_tls) {
21+
// The SSL context created via boost::asio::ssl::context uses the underlying BoringSSL
22+
// implementation (when linked with WebRTC or other BoringSSL-based libraries). BoringSSL is
23+
// not a drop-in replacement for OpenSSL and does not implement all OpenSSL APIs. As a
24+
// result, certain methods may be unsupported or behave differently.
25+
// Ensure that only compatible OpenSSL APIs are used when BoringSSL is present.
26+
DEBUG_PRINT("Using TLS WebSocket, SSL version: %s", OpenSSL_version(OPENSSL_VERSION));
2827

29-
nlohmann::json canditateObj = nlohmann::json::parse(canditateInit);
30-
std::string sdp_mid = canditateObj["sdpMid"];
31-
int sdp_mline_index = canditateObj["sdpMLineIndex"];
32-
std::string candidate = canditateObj["candidate"];
33-
DEBUG_PRINT("Received remote ICE: %s, %d, %s", sdp_mid.c_str(), sdp_mline_index,
34-
candidate.c_str());
28+
ssl::context ctx(ssl::context::tls);
29+
ctx.set_default_verify_paths();
30+
ctx.set_verify_mode(ssl::verify_peer);
3531

36-
if (target == "PUBLISHER") {
37-
pub_peer_->SetRemoteIce(sdp_mid, sdp_mline_index, candidate);
38-
} else if (target == "SUBSCRIBER") {
39-
sub_peer_->SetRemoteIce(sdp_mid, sdp_mline_index, candidate);
32+
return websocket::stream<ssl::stream<tcp::socket>>(net::make_strand(ioc), ctx);
33+
} else {
34+
return websocket::stream<tcp::socket>(net::make_strand(ioc));
4035
}
4136
}
4237

4338
void WebsocketService::Connect() {
39+
auto port = args_.use_tls ? 443 : 80;
40+
INFO_PRINT("Connect to WebSocket %s:%d", args_.ws_host.c_str(), port);
41+
4442
resolver_.async_resolve(
45-
args_.ws_host, std::to_string(args_.ws_port),
43+
args_.ws_host, std::to_string(port),
4644
[this](boost::system::error_code ec, tcp::resolver::results_type results) {
4745
OnResolve(ec, results);
4846
});
4947
}
5048

5149
void WebsocketService::Disconnect() {
52-
if (ws_.is_open()) {
53-
ws_.async_close(websocket::close_code::normal, [this](boost::system::error_code ec) {
54-
if (ec) {
55-
ERROR_PRINT("Close Error: %s", ec.message().c_str());
50+
std::visit(
51+
[](auto &ws) {
52+
if (ws.is_open()) {
53+
ws.async_close(websocket::close_code::normal, [](boost::system::error_code ec) {
54+
if (ec) {
55+
ERROR_PRINT("Close Error: %s", ec.message().c_str());
56+
} else {
57+
INFO_PRINT("WebSocket Closed");
58+
}
59+
});
5660
} else {
57-
INFO_PRINT("WebSocket Closed");
61+
INFO_PRINT("WebSocket already closed");
5862
}
59-
});
60-
} else {
61-
INFO_PRINT("WebSocket already closed");
62-
}
63+
},
64+
ws_);
6365
}
6466

65-
void WebsocketService::OnResolve(boost::system::error_code ec,
66-
tcp::resolver::results_type results) {
67-
if (!ec) {
68-
net::async_connect(ws_.next_layer(), results,
69-
[this](boost::system::error_code ec, tcp::endpoint) {
70-
OnConnect(ec);
71-
});
72-
} else {
67+
void WebsocketService::OnResolve(beast::error_code ec, tcp::resolver::results_type results) {
68+
if (ec) {
7369
ERROR_PRINT("Failed to resolve: %s", ec.message().c_str());
7470
return;
7571
}
72+
73+
std::visit(
74+
[this, results](auto &ws) {
75+
net::async_connect(beast::get_lowest_layer(ws), results,
76+
[this, &ws](boost::system::error_code ec, tcp::endpoint) {
77+
OnConnect(ec);
78+
});
79+
},
80+
ws_);
7681
}
7782

78-
void WebsocketService::OnConnect(boost::system::error_code ec) {
79-
if (!ec) {
80-
std::string target = "/rtc?token=" + args_.ws_token;
81-
ws_.async_handshake(args_.ws_host, target, [this](boost::system::error_code ec) {
82-
OnHandshake(ec);
83-
});
84-
} else {
85-
ERROR_PRINT("Connection Error: %s", ec.message().c_str());
83+
void WebsocketService::OnConnect(beast::error_code ec) {
84+
if (ec) {
85+
ERROR_PRINT("Failed to connect: %s", ec.message().c_str());
86+
return;
8687
}
88+
89+
std::visit(
90+
[this](auto &ws) {
91+
OnHandshake(ws);
92+
},
93+
ws_);
8794
}
8895

89-
void WebsocketService::OnHandshake(boost::system::error_code ec) {
90-
if (!ec) {
91-
INFO_PRINT("WebSocket is connected!");
92-
Read();
93-
} else {
94-
ERROR_PRINT("Handshake Error: %s", ec.message().c_str());
96+
void WebsocketService::OnHandshake(websocket::stream<tcp::socket> &ws) {
97+
std::string target = "/rtc?token=" + args_.ws_token;
98+
ws.async_handshake(args_.ws_host, target, [this](boost::system::error_code ec) {
99+
OnHandshake(ec);
100+
});
101+
}
102+
103+
void WebsocketService::OnHandshake(websocket::stream<ssl::stream<tcp::socket>> &ws) {
104+
ws.next_layer().async_handshake(
105+
ssl::stream_base::client, [this, &ws](boost::system::error_code ec) {
106+
if (ec) {
107+
ERROR_PRINT("Failed to tls handshake: %s", ec.message().c_str());
108+
}
109+
std::string target = "/rtc?token=" + args_.ws_token;
110+
ws.async_handshake(args_.ws_host, target, [this](boost::system::error_code ec) {
111+
OnHandshake(ec);
112+
});
113+
});
114+
}
115+
116+
void WebsocketService::OnHandshake(beast::error_code ec) {
117+
if (ec) {
118+
ERROR_PRINT("Failed to handshake: %s", ec.message().c_str());
119+
return;
95120
}
121+
122+
Read();
96123
}
97124

98125
void WebsocketService::Read() {
99-
ws_.async_read(buffer_, [this](boost::system::error_code ec, std::size_t bytes_transferred) {
100-
if (!ec) {
101-
std::string req = beast::buffers_to_string(buffer_.data());
102-
OnMessage(req);
103-
buffer_.consume(bytes_transferred);
104-
Read();
105-
} else {
106-
ERROR_PRINT("Read Error: %s", ec.message().c_str());
107-
Disconnect();
108-
}
109-
});
126+
std::visit(
127+
[this](auto &ws) {
128+
ws.async_read(buffer_,
129+
[this](boost::system::error_code ec, std::size_t bytes_transferred) {
130+
if (ec) {
131+
ERROR_PRINT("Failed to read: %s", ec.message().c_str());
132+
Disconnect();
133+
}
134+
std::string req = beast::buffers_to_string(buffer_.data());
135+
OnMessage(req);
136+
buffer_.consume(bytes_transferred);
137+
Read();
138+
});
139+
},
140+
ws_);
110141
}
111142

112143
void WebsocketService::OnMessage(const std::string &req) {
@@ -163,6 +194,25 @@ void WebsocketService::OnMessage(const std::string &req) {
163194
}
164195
}
165196

197+
void WebsocketService::OnRemoteIce(const std::string &message) {
198+
nlohmann::json res = nlohmann::json::parse(message);
199+
std::string target = res["target"];
200+
std::string canditateInit = res["candidateInit"];
201+
202+
nlohmann::json canditateObj = nlohmann::json::parse(canditateInit);
203+
std::string sdp_mid = canditateObj["sdpMid"];
204+
int sdp_mline_index = canditateObj["sdpMLineIndex"];
205+
std::string candidate = canditateObj["candidate"];
206+
DEBUG_PRINT("Received remote ICE: %s, %d, %s", sdp_mid.c_str(), sdp_mline_index,
207+
candidate.c_str());
208+
209+
if (target == "PUBLISHER") {
210+
pub_peer_->SetRemoteIce(sdp_mid, sdp_mline_index, candidate);
211+
} else if (target == "SUBSCRIBER") {
212+
sub_peer_->SetRemoteIce(sdp_mid, sdp_mline_index, candidate);
213+
}
214+
}
215+
166216
void WebsocketService::Write(const std::string &action, const std::string &message) {
167217
nlohmann::json request_json;
168218
request_json["action"] = action;
@@ -182,17 +232,22 @@ void WebsocketService::DoWrite() {
182232
if (write_queue_.empty())
183233
return;
184234

185-
ws_.async_write(boost::asio::buffer(write_queue_.front()),
186-
[this](boost::system::error_code ec, std::size_t bytes_transferred) {
187-
std::lock_guard<std::mutex> lock(write_mutex_);
188-
if (!ec) {
189-
write_queue_.pop_front();
190-
if (!write_queue_.empty()) {
191-
DoWrite();
192-
}
193-
} else {
194-
ERROR_PRINT("Write Error: %s", ec.message().c_str());
195-
Disconnect();
196-
}
197-
});
235+
std::visit(
236+
[this](auto &ws) {
237+
ws.async_write(net::buffer(write_queue_.front()),
238+
[this](boost::system::error_code ec, std::size_t bytes_transferred) {
239+
std::lock_guard<std::mutex> lock(write_mutex_);
240+
if (ec) {
241+
ERROR_PRINT("Failed to write: %s", ec.message().c_str());
242+
Disconnect();
243+
}
244+
245+
write_queue_.pop_front();
246+
247+
if (!write_queue_.empty()) {
248+
DoWrite();
249+
}
250+
});
251+
},
252+
ws_);
198253
}

0 commit comments

Comments
 (0)