Branch data Line data Source code
1 : : /*
2 : : * Copyright 2025 Jinwoo Sung
3 : : *
4 : : * Licensed under the Apache License, Version 2.0 (the "License");
5 : : * you may not use this file except in compliance with the License.
6 : : * You may obtain a copy of the License at
7 : : *
8 : : * http://www.apache.org/licenses/LICENSE-2.0
9 : : *
10 : : * Unless required by applicable law or agreed to in writing, software
11 : : * distributed under the License is distributed on an "AS IS" BASIS,
12 : : * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 : : * See the License for the specific language governing permissions and
14 : : * limitations under the License.
15 : : */
16 : :
17 : : #include "wirestead/builder/udp_builder.hpp"
18 : :
19 : : #include <boost/asio/io_context.hpp>
20 : : #include <boost/asio/ip/address.hpp>
21 : :
22 : : #include "wirestead/builder/auto_initializer.hpp"
23 : : #include "wirestead/config/validation.hpp"
24 : : #include "wirestead/diagnostics/exceptions.hpp"
25 : : #include "wirestead/util/input_validator.hpp"
26 : :
27 : : namespace wirestead {
28 : : namespace builder {
29 : :
30 : : // UdpClientBuilder implementation
31 : :
32 : 1 : UdpClientBuilder::UdpClientBuilder() : UdpClientBuilder(0) {}
33 : :
34 : 10 : UdpClientBuilder::UdpClientBuilder(uint16_t local_port)
35 : 10 : : local_port_(local_port),
36 : 10 : bind_address_("0.0.0.0"),
37 : 20 : remote_host_(""),
38 : 10 : remote_port_(0),
39 : 10 : auto_start_(false),
40 : 10 : independent_context_(false),
41 : 10 : enable_broadcast_(false),
42 : 10 : reuse_address_(false),
43 : 10 : send_buffer_size_(0),
44 : 20 : receive_buffer_size_(0) {
45 : : // Ensure background IO service is running
46 : 10 : AutoInitializer::ensure_io_context_running();
47 : 10 : }
48 : :
49 : 7 : std::unique_ptr<wrapper::UdpClient> UdpClientBuilder::build() {
50 : 7 : std::unique_ptr<wrapper::UdpClient> client;
51 : 7 : config::UdpConfig cfg;
52 : 7 : cfg.bind_address = bind_address_;
53 : 7 : cfg.local_port = local_port_;
54 [ + + ]: 7 : if (!remote_host_.empty()) {
55 : 4 : cfg.remote_address = remote_host_;
56 : 4 : cfg.remote_port = remote_port_;
57 : : }
58 : 7 : cfg.enable_broadcast = enable_broadcast_;
59 : 7 : cfg.reuse_address = reuse_address_;
60 : 7 : cfg.multicast_group = multicast_group_;
61 : 7 : cfg.multicast_interface = multicast_interface_;
62 : 7 : cfg.send_buffer_size = send_buffer_size_;
63 : 7 : cfg.receive_buffer_size = receive_buffer_size_;
64 : :
65 [ + + ]: 7 : if (independent_context_) {
66 : 2 : client = std::make_unique<wrapper::UdpClient>(cfg, std::make_shared<boost::asio::io_context>());
67 : 2 : client->manage_external_context(true);
68 : : } else {
69 : 5 : client = std::make_unique<wrapper::UdpClient>(cfg);
70 : : }
71 : :
72 [ + + + - : 7 : if (this->on_data_) client->on_data(this->on_data_);
+ - ]
73 [ + + + - : 7 : if (this->on_data_batch_) client->on_data_batch(this->on_data_batch_);
+ - ]
74 [ + + + - : 7 : if (this->on_connect_) client->on_connect(this->on_connect_);
+ - ]
75 [ + + + - : 7 : if (this->on_disconnect_) client->on_disconnect(this->on_disconnect_);
+ - ]
76 [ + + + - : 7 : if (this->on_error_) client->on_error(this->on_error_);
+ - ]
77 [ + + + - : 7 : if (this->on_backpressure_) client->on_backpressure(this->on_backpressure_);
+ - ]
78 : :
79 [ + + + - ]: 7 : if (this->bp_strategy_set_) client->backpressure_strategy(this->bp_strategy_);
80 : 7 : client->backpressure_threshold(this->get_effective_backpressure_threshold());
81 : :
82 [ + + ]: 7 : if (this->framer_factory_) {
83 : 3 : client->framer(this->framer_factory_());
84 : : }
85 [ + + ]: 7 : if (this->on_message_) {
86 : 2 : client->on_message(this->on_message_);
87 : : }
88 [ + + ]: 7 : if (this->on_message_batch_) {
89 : 1 : client->on_message_batch(this->on_message_batch_);
90 : : }
91 : :
92 [ - + ]: 7 : if (auto_start_) {
93 : 0 : client->auto_start(true);
94 : : }
95 : :
96 : 14 : return client;
97 : 7 : }
98 : :
99 : 4 : UdpClientBuilder& UdpClientBuilder::auto_start(bool auto_start) {
100 : 4 : auto_start_ = auto_start;
101 : 4 : return *this;
102 : : }
103 : :
104 : 3 : UdpClientBuilder& UdpClientBuilder::local_port(uint16_t port) {
105 : 3 : local_port_ = port;
106 : 3 : return *this;
107 : : }
108 : :
109 : 2 : UdpClientBuilder& UdpClientBuilder::bind_address(const std::string& address) {
110 [ - + - - ]: 2 : config::detail::require(util::InputValidator::is_valid_ipv4(address) || util::InputValidator::is_valid_ipv6(address),
111 : : "invalid bind address");
112 : 2 : bind_address_ = address;
113 : 2 : return *this;
114 : : }
115 : :
116 : 6 : UdpClientBuilder& UdpClientBuilder::remote_endpoint(const std::string& host, uint16_t port) {
117 : 6 : boost::system::error_code ec;
118 : 6 : boost::asio::ip::make_address(host, ec);
119 [ + + ]: 6 : if (ec) {
120 : 5 : throw diagnostics::BuilderException("Invalid remote address: " + host, "udp");
121 : : }
122 [ + + ]: 5 : config::detail::require(port != 0, "invalid remote port");
123 : 4 : remote_host_ = host;
124 : 4 : remote_port_ = port;
125 : 4 : return *this;
126 : : }
127 : :
128 : 4 : UdpClientBuilder& UdpClientBuilder::broadcast(bool enable) {
129 : 4 : enable_broadcast_ = enable;
130 : 4 : return *this;
131 : : }
132 : :
133 : 2 : UdpClientBuilder& UdpClientBuilder::multicast_group(const std::string& group, const std::string& interface_address) {
134 : 2 : config::detail::require(config::is_multicast_address(group), "invalid multicast group");
135 [ + - - + ]: 1 : config::detail::require(interface_address.empty() || util::InputValidator::is_valid_ipv4(interface_address),
136 : : "invalid multicast interface");
137 : 0 : multicast_group_ = group;
138 [ # # ]: 0 : if (interface_address.empty()) {
139 : 0 : multicast_interface_.reset();
140 : : } else {
141 : 0 : multicast_interface_ = interface_address;
142 : : }
143 : 0 : return *this;
144 : : }
145 : :
146 : 4 : UdpClientBuilder& UdpClientBuilder::reuse_address(bool enable) {
147 : 4 : reuse_address_ = enable;
148 : 4 : return *this;
149 : : }
150 : :
151 : 3 : UdpClientBuilder& UdpClientBuilder::independent_context(bool use_independent) {
152 : 3 : independent_context_ = use_independent;
153 : 3 : return *this;
154 : : }
155 : :
156 : 2 : UdpClientBuilder& UdpClientBuilder::send_buffer_size(size_t bytes) {
157 [ + - ]: 2 : if (bytes != 0)
158 : 2 : config::detail::range(bytes, base::constants::MIN_SOCKET_BUFFER_SIZE, base::constants::MAX_SOCKET_BUFFER_SIZE,
159 : : "invalid send_buffer_size");
160 : 1 : send_buffer_size_ = bytes;
161 : 1 : return *this;
162 : : }
163 : :
164 : 1 : UdpClientBuilder& UdpClientBuilder::receive_buffer_size(size_t bytes) {
165 [ + - ]: 1 : if (bytes != 0)
166 : 1 : config::detail::range(bytes, base::constants::MIN_SOCKET_BUFFER_SIZE, base::constants::MAX_SOCKET_BUFFER_SIZE,
167 : : "invalid receive_buffer_size");
168 : 1 : receive_buffer_size_ = bytes;
169 : 1 : return *this;
170 : : }
171 : :
172 : : // UdpServerBuilder implementation
173 : :
174 : 1 : UdpServerBuilder::UdpServerBuilder() : UdpServerBuilder(0) {}
175 : :
176 : 5 : UdpServerBuilder::UdpServerBuilder(uint16_t local_port)
177 : 5 : : local_port_(local_port),
178 : 5 : bind_address_("0.0.0.0"),
179 : 5 : auto_start_(false),
180 : 5 : independent_context_(false),
181 : 5 : enable_broadcast_(false),
182 : 5 : reuse_address_(false),
183 : 5 : send_buffer_size_(0),
184 : 10 : receive_buffer_size_(0) {
185 : : // Ensure background IO service is running
186 : 5 : AutoInitializer::ensure_io_context_running();
187 : 5 : }
188 : :
189 : 4 : std::unique_ptr<wrapper::UdpServer> UdpServerBuilder::build() {
190 : 4 : std::unique_ptr<wrapper::UdpServer> server;
191 : 4 : config::UdpConfig cfg;
192 : 4 : cfg.bind_address = bind_address_;
193 : 4 : cfg.local_port = local_port_;
194 : 4 : cfg.enable_broadcast = enable_broadcast_;
195 : 4 : cfg.reuse_address = reuse_address_;
196 : 4 : cfg.multicast_group = multicast_group_;
197 : 4 : cfg.multicast_interface = multicast_interface_;
198 : 4 : cfg.send_buffer_size = send_buffer_size_;
199 : 4 : cfg.receive_buffer_size = receive_buffer_size_;
200 : :
201 [ + + ]: 4 : if (independent_context_) {
202 : 2 : server = std::make_unique<wrapper::UdpServer>(cfg, std::make_shared<boost::asio::io_context>());
203 : 2 : server->manage_external_context(true);
204 : : } else {
205 : 2 : server = std::make_unique<wrapper::UdpServer>(cfg);
206 : : }
207 : :
208 [ + + + - : 4 : if (this->on_data_) server->on_data(this->on_data_);
+ - ]
209 [ + + + - : 4 : if (this->on_data_batch_) server->on_data_batch(this->on_data_batch_);
+ - ]
210 [ + + + - : 4 : if (this->on_connect_) server->on_connect(this->on_connect_);
+ - ]
211 [ + + + - : 4 : if (this->on_disconnect_) server->on_disconnect(this->on_disconnect_);
+ - ]
212 [ + + + - : 4 : if (this->on_error_) server->on_error(this->on_error_);
+ - ]
213 [ + + + - : 4 : if (this->on_backpressure_) server->on_backpressure(this->on_backpressure_);
+ - ]
214 : :
215 [ + + + - ]: 4 : if (this->bp_strategy_set_) server->backpressure_strategy(this->bp_strategy_);
216 : 4 : server->backpressure_threshold(this->get_effective_backpressure_threshold());
217 : :
218 [ + + ]: 4 : if (this->framer_factory_) {
219 : 3 : server->framer(this->framer_factory_);
220 : : }
221 [ + + ]: 4 : if (this->on_message_) {
222 : 2 : server->on_message(this->on_message_);
223 : : }
224 [ + + ]: 4 : if (this->on_message_batch_) {
225 : 1 : server->on_message_batch(this->on_message_batch_);
226 : : }
227 : :
228 [ + + ]: 4 : if (client_limit_enabled_) {
229 : 2 : server->max_clients(max_clients_);
230 : : }
231 [ + + ]: 4 : if (idle_timeout_set_) {
232 : 2 : server->idle_timeout(idle_timeout_);
233 : : }
234 : :
235 [ - + ]: 4 : if (auto_start_) {
236 : 0 : server->auto_start(true);
237 : : }
238 : :
239 : 8 : return server;
240 : 4 : }
241 : :
242 : 4 : UdpServerBuilder& UdpServerBuilder::auto_start(bool auto_start) {
243 : 4 : auto_start_ = auto_start;
244 : 4 : return *this;
245 : : }
246 : :
247 : 3 : UdpServerBuilder& UdpServerBuilder::local_port(uint16_t port) {
248 : 3 : local_port_ = port;
249 : 3 : return *this;
250 : : }
251 : :
252 : 2 : UdpServerBuilder& UdpServerBuilder::bind_address(const std::string& address) {
253 [ - + - - ]: 2 : config::detail::require(util::InputValidator::is_valid_ipv4(address) || util::InputValidator::is_valid_ipv6(address),
254 : : "invalid bind address");
255 : 2 : bind_address_ = address;
256 : 2 : return *this;
257 : : }
258 : :
259 : 2 : UdpServerBuilder& UdpServerBuilder::max_clients(uint32_t max) {
260 : 2 : config::detail::range(max, 0, base::constants::MAX_MAX_CONNECTIONS, "invalid client limit");
261 : 2 : max_clients_ = max;
262 : 2 : client_limit_enabled_ = true;
263 : 2 : return *this;
264 : : }
265 : :
266 : 4 : UdpServerBuilder& UdpServerBuilder::broadcast(bool enable) {
267 : 4 : enable_broadcast_ = enable;
268 : 4 : return *this;
269 : : }
270 : :
271 : 0 : UdpServerBuilder& UdpServerBuilder::multicast_group(const std::string& group, const std::string& interface_address) {
272 : 0 : config::detail::require(config::is_multicast_address(group), "invalid multicast group");
273 [ # # # # ]: 0 : config::detail::require(interface_address.empty() || util::InputValidator::is_valid_ipv4(interface_address),
274 : : "invalid multicast interface");
275 : 0 : multicast_group_ = group;
276 [ # # ]: 0 : if (interface_address.empty()) {
277 : 0 : multicast_interface_.reset();
278 : : } else {
279 : 0 : multicast_interface_ = interface_address;
280 : : }
281 : 0 : return *this;
282 : : }
283 : :
284 : 4 : UdpServerBuilder& UdpServerBuilder::reuse_address(bool enable) {
285 : 4 : reuse_address_ = enable;
286 : 4 : return *this;
287 : : }
288 : :
289 : 3 : UdpServerBuilder& UdpServerBuilder::independent_context(bool use_independent) {
290 : 3 : independent_context_ = use_independent;
291 : 3 : return *this;
292 : : }
293 : :
294 : 2 : UdpServerBuilder& UdpServerBuilder::idle_timeout(std::chrono::milliseconds timeout) {
295 : 2 : config::detail::duration(timeout, base::constants::MIN_IDLE_TIMEOUT_MS, base::constants::MAX_IDLE_TIMEOUT_MS, true,
296 : : "invalid idle_timeout");
297 : 2 : idle_timeout_ = timeout;
298 : 2 : idle_timeout_set_ = true;
299 : 2 : return *this;
300 : : }
301 : :
302 : 1 : UdpServerBuilder& UdpServerBuilder::send_buffer_size(size_t bytes) {
303 [ + - ]: 1 : if (bytes != 0)
304 : 1 : config::detail::range(bytes, base::constants::MIN_SOCKET_BUFFER_SIZE, base::constants::MAX_SOCKET_BUFFER_SIZE,
305 : : "invalid send_buffer_size");
306 : 1 : send_buffer_size_ = bytes;
307 : 1 : return *this;
308 : : }
309 : :
310 : 1 : UdpServerBuilder& UdpServerBuilder::receive_buffer_size(size_t bytes) {
311 [ + - ]: 1 : if (bytes != 0)
312 : 1 : config::detail::range(bytes, base::constants::MIN_SOCKET_BUFFER_SIZE, base::constants::MAX_SOCKET_BUFFER_SIZE,
313 : : "invalid receive_buffer_size");
314 : 1 : receive_buffer_size_ = bytes;
315 : 1 : return *this;
316 : : }
317 : :
318 : : // Explicit template instantiations
319 : :
320 : : } // namespace builder
321 : : } // namespace wirestead
|
