Branch data Line data Source code
1 : : /*
2 : : * Copyright (C) 2026 by Thun Lu. All rights reserved.
3 : : * Author: Thun Lu <thun.lu@zohomail.cn>
4 : : * Repo: https://github.com/thun-res/vlink
5 : : * _ __ __ _ __
6 : : * | | / / / / (_) ____ / /__
7 : : * | | / / / / / / / __ \ / //_/
8 : : * | |/ / / /___ / / / / / / / ,<
9 : : * |___/ /_____/ /_/ /_/ /_/ /_/|_|
10 : : *
11 : : * Licensed under the Apache License, Version 2.0 (the "License");
12 : : * you may not use this file except in compliance with the License.
13 : : * You may obtain a copy of the License at
14 : : *
15 : : * http://www.apache.org/licenses/LICENSE-2.0
16 : : *
17 : : * Unless required by applicable law or agreed to in writing, software
18 : : * distributed under the License is distributed on an "AS IS" BASIS,
19 : : * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
20 : : * See the License for the specific language governing permissions and
21 : : * limitations under the License.
22 : : */
23 : :
24 : : #include "./ddsc_factory.hpp"
25 : :
26 : : #include <dds/ddsi/ddsi_config.h>
27 : :
28 : : #include <charconv>
29 : : #include <memory>
30 : : #include <string>
31 : : #include <utility>
32 : : #include <vector>
33 : :
34 : : #include "./base/helpers.h"
35 : : #include "./base/utils.h"
36 : : #include "./ddsc_qos.hpp"
37 : : #include "./extension/qos_profile.h"
38 : : #include "./impl/ssl_options.h"
39 : :
40 : : namespace vlink {
41 : :
42 : : // DdscFactory
43 [ + - ]: 8 : DdscFactory::DdscFactory() {
44 : 8 : Bytes::init_memory_pool();
45 : :
46 [ - + ]: 8 : if VUNLIKELY (DdscConf::get_thread_count() != 1) {
47 [ # # # # ]: 0 : VLOG_W("DdscFactory: Ddsc does not support setting thread count.");
48 : : }
49 : :
50 [ + + ]: 136 : for (const auto& [name, qos] : QosProfile::get_available_qos_map()) {
51 [ + - ]: 128 : DdscConf::register_qos_internal(name, qos);
52 : : }
53 : :
54 [ + - + - ]: 8 : if (process_cyclone_dds_uri().empty()) {
55 [ + - + - ]: 16 : std::string dds_debug_str = Utils::get_env("VLINK_DDS_DEBUG");
56 : :
57 [ - + ]: 8 : if (dds_debug_str == "1") {
58 [ # # ]: 0 : dds_set_log_mask(DDS_LC_ALL);
59 : : } else {
60 [ + - ]: 8 : dds_set_log_mask(DDS_LC_FATAL);
61 : : }
62 : 8 : }
63 : :
64 [ + - + - ]: 16 : std::string default_event_qos_str = Utils::get_env("VLINK_DDS_EVENT_QOS");
65 [ + - + - ]: 16 : std::string default_method_qos_str = Utils::get_env("VLINK_DDS_METHOD_QOS");
66 [ + - + - ]: 16 : std::string default_field_qos_str = Utils::get_env("VLINK_DDS_FIELD_QOS");
67 : :
68 [ + - ]: 8 : if (default_event_qos_str.empty()) {
69 : 8 : default_event_qos_ = QosProfile::kEvent;
70 : : } else {
71 [ # # ]: 0 : default_event_qos_ = DdscConf::find_qos(default_event_qos_str);
72 : : }
73 : :
74 [ + - ]: 8 : if (default_method_qos_str.empty()) {
75 : 8 : default_method_qos_ = QosProfile::kMethod;
76 : : } else {
77 [ # # ]: 0 : default_method_qos_ = DdscConf::find_qos(default_method_qos_str);
78 : : }
79 : :
80 [ + - ]: 8 : if (default_field_qos_str.empty()) {
81 : 8 : default_field_qos_ = QosProfile::kField;
82 : : } else {
83 [ # # ]: 0 : default_field_qos_ = DdscConf::find_qos(default_field_qos_str);
84 : : }
85 : 8 : }
86 : :
87 : 8 : DdscFactory::~DdscFactory() = default;
88 : :
89 : 191 : std::shared_ptr<ddsc::DomainParticipant> DdscFactory::create_participant(uint8_t type, const DdscConf& conf,
90 : : const Conf::PropertiesMap& properties) {
91 [ + + + - : 191 : static auto& factory = DdscFactory::get();
+ - - - ]
92 : :
93 [ + - ]: 191 : const auto& id = std::make_tuple(type, conf.domain, properties);
94 [ + - ]: 191 : std::unique_lock lock(factory.mtx_);
95 [ + - ]: 191 : std::shared_ptr<ddsc::DomainParticipant> part = get_weak_ptr(factory.part_map_, id).lock();
96 : :
97 [ + + ]: 191 : if (!part) {
98 [ + - ]: 88 : factory.part_map_.erase(id);
99 : :
100 [ + - ]: 88 : dds_qos_t* dds_qos = dds_create_qos();
101 : :
102 [ + - ]: 88 : set_participant_qos(conf.domain, dds_qos, properties);
103 : :
104 : 88 : bool has_domain_ref = false;
105 [ + - ]: 88 : auto domain_iter = factory.domain_map_.find(conf.domain);
106 : :
107 [ + - ]: 88 : if (domain_iter != factory.domain_map_.end()) {
108 : 88 : ++domain_iter->second.ref_count;
109 : 88 : has_domain_ref = true;
110 : : }
111 : :
112 [ + - + - ]: 88 : auto* ptr = new ddsc::DomainParticipant(conf.domain, dds_qos);
113 : :
114 [ + - - + : 88 : if VUNLIKELY (!ptr || ptr->entity <= 0) {
- + ]
115 [ # # # # ]: 0 : VLOG_E("DdscFactory: Failed to create participant.");
116 [ # # ]: 0 : delete ptr;
117 [ # # ]: 0 : dds_delete_qos(dds_qos);
118 : :
119 [ # # ]: 0 : if (has_domain_ref) {
120 [ # # ]: 0 : auto iter = factory.domain_map_.find(conf.domain);
121 [ # # # # : 0 : if VLIKELY (iter != factory.domain_map_.end() && iter->second.ref_count > 0) {
# # ]
122 : 0 : --iter->second.ref_count;
123 : : }
124 : : }
125 : :
126 : 0 : return nullptr;
127 : : }
128 : :
129 [ + - ]: 176 : part = std::shared_ptr<ddsc::DomainParticipant>(
130 [ + - ]: 88 : ptr, [id, domain = conf.domain, has_domain_ref](ddsc::DomainParticipant* part) {
131 : : {
132 [ + - ]: 88 : std::lock_guard lock(factory.mtx_);
133 : :
134 [ + - + - : 88 : if (auto iter = factory.part_map_.find(id); iter != factory.part_map_.end() && iter->second.expired()) {
+ - + - ]
135 [ + - ]: 88 : factory.part_map_.erase(iter);
136 : : }
137 : :
138 [ + - ]: 88 : delete part;
139 : :
140 [ - + ]: 88 : if (!has_domain_ref) {
141 : 0 : return;
142 : : }
143 : :
144 [ + - ]: 88 : auto iter = factory.domain_map_.find(domain);
145 : :
146 [ + - + - : 88 : if VLIKELY (iter != factory.domain_map_.end() && iter->second.ref_count > 0) {
+ - ]
147 : 88 : --iter->second.ref_count;
148 : :
149 [ + + ]: 88 : if (iter->second.ref_count == 0) {
150 [ + - ]: 84 : dds_delete(iter->second.entity);
151 [ + - ]: 84 : factory.domain_map_.erase(iter);
152 : : }
153 : : }
154 [ + - ]: 88 : }
155 : 88 : });
156 : :
157 [ + - ]: 88 : factory.part_map_.emplace(id, part);
158 : :
159 [ + - ]: 88 : dds_delete_qos(dds_qos);
160 : : }
161 : :
162 : 191 : return part;
163 : 191 : }
164 : :
165 : 234 : std::shared_ptr<ddsc::Topic> DdscFactory::create_topic(uint8_t type, const DdscConf& conf,
166 : : ddsc::DomainParticipant* part, std::string topic) {
167 [ + + + - : 234 : static auto& factory = DdscFactory::get();
+ - - - ]
168 : :
169 [ + + ]: 234 : if (topic.empty()) {
170 [ + - ]: 148 : topic = conf.topic;
171 : : }
172 : :
173 [ - + ]: 234 : if VUNLIKELY (!part) {
174 [ # # # # ]: 0 : VLOG_E("DdscFactory: Cannot create topic without participant.");
175 : 0 : return nullptr;
176 : : }
177 : :
178 [ + - ]: 234 : const auto& id = std::make_tuple(type, conf.domain, topic, part);
179 [ + - ]: 234 : std::unique_lock lock(factory.mtx_);
180 [ + - ]: 234 : std::shared_ptr<ddsc::Topic> dds_topic = get_weak_ptr(factory.topic_map_, id).lock();
181 : :
182 [ + + ]: 234 : if (!dds_topic) {
183 [ + - ]: 129 : lock.unlock();
184 [ + - + - : 129 : auto* ptr = new ddsc::Topic(part->entity, topic);
+ - ]
185 : :
186 [ + - - + : 129 : if VUNLIKELY (!ptr || ptr->entity <= 0) {
- + ]
187 [ # # # # ]: 0 : VLOG_E("DdscFactory: Failed to create topic: ", topic, ".");
188 [ # # ]: 0 : delete ptr;
189 : 0 : return nullptr;
190 : : }
191 : :
192 [ + - + - ]: 258 : dds_topic = std::shared_ptr<ddsc::Topic>(ptr, [id](ddsc::Topic* topic) {
193 : : {
194 [ + - ]: 129 : std::lock_guard lock(factory.mtx_);
195 [ + - ]: 129 : auto iter = factory.topic_map_.find(id);
196 : :
197 [ + - + - : 129 : if (iter != factory.topic_map_.end() && iter->second.expired()) {
+ - ]
198 [ + - ]: 129 : factory.topic_map_.erase(iter);
199 : : }
200 : 129 : }
201 [ + - ]: 129 : delete topic;
202 : 258 : });
203 : :
204 [ + - ]: 129 : lock.lock();
205 : :
206 [ + - ]: 129 : auto [iter, inserted] = factory.topic_map_.emplace(id, dds_topic);
207 : :
208 [ - + ]: 129 : if (!inserted) {
209 : 0 : auto inserted_topic = iter->second.lock();
210 [ # # ]: 0 : if VLIKELY (inserted_topic) {
211 [ # # ]: 0 : lock.unlock();
212 : 0 : dds_topic = std::move(inserted_topic);
213 : : } else {
214 : 0 : iter->second = dds_topic;
215 : : }
216 : 0 : }
217 : : }
218 : :
219 : 234 : return dds_topic;
220 : 234 : }
221 : :
222 : 43 : std::pair<std::shared_ptr<ddsc::Topic>, std::shared_ptr<ddsc::Topic> > DdscFactory::create_method_topic(
223 : : uint8_t type, const DdscConf& conf, ddsc::DomainParticipant* part) {
224 [ + - ]: 43 : const std::string& resp_topic = conf.topic + DdscConf::kRespSuffix;
225 : :
226 [ + - - + : 43 : if VUNLIKELY (conf.topic.empty() || resp_topic.empty()) {
- + ]
227 [ # # # # ]: 0 : VLOG_F("DdscFactory: Method conf topic error.");
228 : : }
229 : :
230 [ - + ]: 43 : if VUNLIKELY (conf.topic == resp_topic) {
231 [ # # # # ]: 0 : VLOG_F("DdscFactory: Method conf topic req and resp cannot be equal.");
232 : : }
233 : :
234 [ + - + - : 86 : return {create_topic(type, conf, part, conf.topic), create_topic(type, conf, part, resp_topic)};
+ - + - ]
235 : 43 : }
236 : :
237 : 116 : std::shared_ptr<ddsc::Publisher> DdscFactory::create_publisher(uint8_t type, const DdscConf& conf,
238 : : ddsc::DomainParticipant* part) {
239 [ + + + - : 116 : static auto& factory = DdscFactory::get();
+ - - - ]
240 : :
241 [ - + ]: 116 : if VUNLIKELY (!part) {
242 [ # # # # ]: 0 : VLOG_E("DdscFactory: Cannot create publisher without participant.");
243 : 0 : return nullptr;
244 : : }
245 : :
246 [ + - ]: 116 : const auto& id = std::make_tuple(type, conf.domain, conf.qos, part);
247 [ + - ]: 116 : std::unique_lock lock(factory.mtx_);
248 [ + - ]: 116 : std::shared_ptr<ddsc::Publisher> publisher = get_weak_ptr(factory.publisher_map_, id).lock();
249 : :
250 [ + + ]: 116 : if (!publisher) {
251 [ + - ]: 101 : lock.unlock();
252 : :
253 [ + - + - ]: 101 : auto* ptr = new ddsc::Publisher(part->entity);
254 : :
255 [ + - - + : 101 : if VUNLIKELY (!ptr || ptr->entity <= 0) {
- + ]
256 [ # # # # ]: 0 : VLOG_E("DdscFactory: Failed to create publisher.");
257 [ # # ]: 0 : delete ptr;
258 : 0 : return nullptr;
259 : : }
260 : :
261 [ + - + - ]: 202 : publisher = std::shared_ptr<ddsc::Publisher>(ptr, [id](ddsc::Publisher* publisher) {
262 : : {
263 [ + - ]: 101 : std::lock_guard lock(factory.mtx_);
264 [ + - ]: 101 : auto iter = factory.publisher_map_.find(id);
265 : :
266 [ + - + - : 101 : if (iter != factory.publisher_map_.end() && iter->second.expired()) {
+ - ]
267 [ + - ]: 101 : factory.publisher_map_.erase(iter);
268 : : }
269 : 101 : }
270 : :
271 [ + - ]: 101 : delete publisher;
272 : 202 : });
273 : :
274 [ + - ]: 101 : lock.lock();
275 : :
276 [ + - ]: 101 : auto [iter, inserted] = factory.publisher_map_.emplace(id, publisher);
277 : :
278 [ - + ]: 101 : if (!inserted) {
279 : 0 : auto inserted_publisher = iter->second.lock();
280 [ # # ]: 0 : if VLIKELY (inserted_publisher) {
281 [ # # ]: 0 : lock.unlock();
282 : 0 : publisher = std::move(inserted_publisher);
283 : : } else {
284 : 0 : iter->second = publisher;
285 : : }
286 : 0 : }
287 : : }
288 : :
289 : 116 : return publisher;
290 : 116 : }
291 : :
292 : 118 : std::shared_ptr<ddsc::Subscriber> DdscFactory::create_subscriber(uint8_t type, const DdscConf& conf,
293 : : ddsc::DomainParticipant* part) {
294 [ + + + - : 118 : static auto& factory = DdscFactory::get();
+ - - - ]
295 : :
296 [ - + ]: 118 : if VUNLIKELY (!part) {
297 [ # # # # ]: 0 : VLOG_E("DdscFactory: Cannot create subscriber without participant.");
298 : 0 : return nullptr;
299 : : }
300 : :
301 [ + - ]: 118 : const auto& id = std::make_tuple(type, conf.domain, conf.qos, part);
302 [ + - ]: 118 : std::unique_lock lock(factory.mtx_);
303 [ + - ]: 118 : std::shared_ptr<ddsc::Subscriber> subscriber = get_weak_ptr(factory.subscriber_map_, id).lock();
304 : :
305 [ + + ]: 118 : if (!subscriber) {
306 [ + - ]: 103 : lock.unlock();
307 : :
308 [ + - + - ]: 103 : auto* ptr = new ddsc::Subscriber(part->entity);
309 : :
310 [ + - - + : 103 : if VUNLIKELY (!ptr || ptr->entity <= 0) {
- + ]
311 [ # # # # ]: 0 : VLOG_E("DdscFactory: Failed to create subscriber.");
312 [ # # ]: 0 : delete ptr;
313 : 0 : return nullptr;
314 : : }
315 : :
316 [ + - + - ]: 206 : subscriber = std::shared_ptr<ddsc::Subscriber>(ptr, [id](ddsc::Subscriber* subscriber) {
317 : : {
318 [ + - ]: 103 : std::lock_guard lock(factory.mtx_);
319 [ + - ]: 103 : auto iter = factory.subscriber_map_.find(id);
320 : :
321 [ + - + - : 103 : if (iter != factory.subscriber_map_.end() && iter->second.expired()) {
+ - ]
322 [ + - ]: 103 : factory.subscriber_map_.erase(iter);
323 : : }
324 : 103 : }
325 : :
326 [ + - ]: 103 : delete subscriber;
327 : 206 : });
328 : :
329 [ + - ]: 103 : lock.lock();
330 : :
331 [ + - ]: 103 : auto [iter, inserted] = factory.subscriber_map_.emplace(id, subscriber);
332 : :
333 [ - + ]: 103 : if (!inserted) {
334 : 0 : auto inserted_subscriber = iter->second.lock();
335 [ # # ]: 0 : if VLIKELY (inserted_subscriber) {
336 [ # # ]: 0 : lock.unlock();
337 : 0 : subscriber = std::move(inserted_subscriber);
338 : : } else {
339 : 0 : iter->second = subscriber;
340 : : }
341 : 0 : }
342 : : }
343 : :
344 : 118 : return subscriber;
345 : 118 : }
346 : :
347 : 116 : std::shared_ptr<ddsc::DataWriter> DdscFactory::create_datawriter(uint8_t type, const DdscConf& conf,
348 : : ddsc::Publisher* publisher, ddsc::Topic* topic,
349 : : dds_listener_t* listener) {
350 [ + + + - : 116 : static auto& factory = DdscFactory::get();
+ - - - ]
351 : :
352 [ + - - + : 116 : if VUNLIKELY (!publisher || !topic) {
- + ]
353 [ # # # # ]: 0 : VLOG_E("DdscFactory: Cannot create datawriter without publisher/topic.");
354 : 0 : return nullptr;
355 : : }
356 : :
357 [ + - ]: 116 : dds_qos_t* dds_qos = dds_create_qos();
358 : :
359 [ + + ]: 116 : if (conf.qos.empty()) {
360 [ + + - + ]: 103 : if ((type & kPublisher) || (type & kSubscriber)) {
361 [ + - ]: 35 : convert_qos(dds_qos, factory.default_event_qos_, conf.depth);
362 [ + + + + ]: 68 : } else if ((type & kClient) || (type & kServer)) {
363 [ + - ]: 50 : convert_qos(dds_qos, factory.default_method_qos_, conf.depth);
364 [ - + - - ]: 18 : } else if ((type & kSetter) || (type & kGetter)) {
365 [ + - ]: 18 : convert_qos(dds_qos, factory.default_field_qos_, conf.depth);
366 : : }
367 : : } else {
368 [ + - + - ]: 13 : convert_qos(dds_qos, DdscConf::find_qos(conf.qos), conf.depth);
369 : : }
370 : :
371 [ + - ]: 116 : auto writer = std::make_shared<ddsc::DataWriter>(publisher->entity, topic->entity, dds_qos, listener);
372 [ + - ]: 116 : dds_delete_qos(dds_qos);
373 : :
374 [ + - - + : 116 : if VUNLIKELY (!writer || writer->entity <= 0) {
- + ]
375 [ # # # # ]: 0 : VLOG_E("DdscFactory: Failed to create datawriter.");
376 : 0 : return nullptr;
377 : : }
378 : :
379 : 116 : return writer;
380 : 116 : }
381 : :
382 : 109 : std::shared_ptr<ddsc::DataReader> DdscFactory::create_datareader(uint8_t type, const DdscConf& conf,
383 : : ddsc::Subscriber* subscriber, ddsc::Topic* topic,
384 : : dds_listener_t* listener) {
385 [ + + + - : 109 : static auto& factory = DdscFactory::get();
+ - - - ]
386 : :
387 [ + - - + : 109 : if VUNLIKELY (!subscriber || !topic) {
- + ]
388 [ # # # # ]: 0 : VLOG_E("DdscFactory: Cannot create datareader without subscriber/topic.");
389 : 0 : return nullptr;
390 : : }
391 : :
392 [ + - ]: 109 : dds_qos_t* dds_qos = dds_create_qos();
393 : :
394 [ + + ]: 109 : if (conf.qos.empty()) {
395 [ + - + + ]: 96 : if ((type & kPublisher) || (type & kSubscriber)) {
396 [ + - ]: 28 : convert_qos(dds_qos, factory.default_event_qos_, conf.depth);
397 [ + + + + ]: 68 : } else if ((type & kClient) || (type & kServer)) {
398 [ + - ]: 47 : convert_qos(dds_qos, factory.default_method_qos_, conf.depth);
399 [ + - + - ]: 21 : } else if ((type & kSetter) || (type & kGetter)) {
400 [ + - ]: 21 : convert_qos(dds_qos, factory.default_field_qos_, conf.depth);
401 : : }
402 : : } else {
403 [ + - + - ]: 13 : convert_qos(dds_qos, DdscConf::find_qos(conf.qos), conf.depth);
404 : : }
405 : :
406 [ + - ]: 109 : auto reader = std::make_shared<ddsc::DataReader>(subscriber->entity, topic->entity, dds_qos, listener);
407 [ + - ]: 109 : dds_delete_qos(dds_qos);
408 : :
409 [ + - - + : 109 : if VUNLIKELY (!reader || reader->entity <= 0) {
- + ]
410 [ # # # # ]: 0 : VLOG_E("DdscFactory: Failed to create datareader.");
411 : 0 : return nullptr;
412 : : }
413 : :
414 : 109 : return reader;
415 : 109 : }
416 : :
417 : 158 : bool DdscFactory::write_data(dds_entity_t entity, const Bytes& bytes, uint64_t id) {
418 : : vlink_BuiltInRaw msg;
419 : :
420 : 158 : msg.id = id;
421 : 158 : msg.data._buffer = const_cast<uint8_t*>(bytes.data());
422 : 158 : msg.data._length = bytes.size();
423 : 158 : msg.data._maximum = bytes.size();
424 : 158 : msg.data._release = false;
425 : :
426 [ + - ]: 158 : auto ret = dds_write(entity, &msg);
427 : :
428 : 158 : return ret >= 0;
429 : : }
430 : :
431 : 356 : bool DdscFactory::take_data(dds_entity_t entity, ReadMessage& msg) {
432 : 356 : auto ret = dds_take_next(entity, &msg.sample, &msg.info);
433 : :
434 [ + + ]: 356 : if (ret == 0) {
435 : 178 : return false;
436 : : }
437 : :
438 [ - + ]: 178 : if (ret == DDS_RETCODE_NO_DATA) {
439 : 0 : return false;
440 : : }
441 : :
442 [ - + ]: 178 : if VUNLIKELY (ret < 0) {
443 [ # # # # ]: 0 : VLOG_E("DdscFactory: Failed to take data.");
444 : :
445 : 0 : return false;
446 : : }
447 : :
448 : 178 : auto* sample = static_cast<vlink_BuiltInRaw*>(msg.sample);
449 : :
450 : 178 : msg.id = sample->id;
451 : :
452 : 178 : msg.bytes = Bytes::shallow_copy(sample->data._buffer, sample->data._length);
453 : :
454 : 178 : msg.timestamp = msg.info.source_timestamp;
455 : :
456 : 178 : msg.guid = msg.info.publication_handle;
457 : :
458 : 178 : return true;
459 : : }
460 : :
461 : 178 : bool DdscFactory::release_data(dds_entity_t entity, ReadMessage& msg) {
462 : 178 : auto ret = dds_return_loan(entity, &msg.sample, 1);
463 : :
464 : 178 : msg.sample = nullptr;
465 : :
466 : 178 : return ret == DDS_RETCODE_OK;
467 : : }
468 : :
469 : 25 : uint64_t DdscFactory::get_guid(const dds_guid_t* guid, uint32_t seq) {
470 : 25 : uint64_t result = 14695981039346656037ULL;
471 : :
472 [ + + ]: 425 : for (const auto value : guid->v) {
473 : 400 : result ^= static_cast<uint64_t>(value);
474 : 400 : result *= 1099511628211ULL;
475 : : }
476 : :
477 [ + + ]: 125 : for (size_t i = 0; i < sizeof(seq); ++i) {
478 : 100 : result ^= static_cast<uint64_t>((seq >> (i * 8)) & 0xFFU);
479 : 100 : result *= 1099511628211ULL;
480 : : }
481 : :
482 : 25 : return result;
483 : : }
484 : :
485 : 9 : int DdscFactory::get_default_domain_id() {
486 [ + - + - ]: 9 : const std::string& domain_str = Utils::get_env("VLINK_DDS_DOMAIN");
487 : 18 : return Helpers::to_int(domain_str, 0);
488 : 9 : }
489 : :
490 : 16 : std::string DdscFactory::process_cyclone_dds_uri() {
491 [ + - + - ]: 32 : std::string cyclone_dds_uri = Utils::get_env("VLINK_CYCLONEDDS_URI");
492 : :
493 [ - + ]: 16 : if (!cyclone_dds_uri.empty()) {
494 [ # # ]: 0 : Utils::set_env("CYCLONEDDS_URI", cyclone_dds_uri);
495 : : }
496 : :
497 : 16 : return cyclone_dds_uri;
498 : 0 : }
499 : :
500 : 88 : void DdscFactory::set_participant_qos(int32_t domain_id, dds_qos_t* dds_qos, const Conf::PropertiesMap& properties) {
501 [ + + + - : 88 : static auto& factory = DdscFactory::get();
+ - - - ]
502 : :
503 [ + + + - : 88 : static const std::string& cyclone_dds_uri = process_cyclone_dds_uri();
+ - - - ]
504 : :
505 [ - + ]: 88 : if (!cyclone_dds_uri.empty()) {
506 : 4 : return;
507 : : }
508 : :
509 [ + + + - : 88 : static const std::string& ip_str = Utils::get_env("VLINK_DDS_IP");
+ - + - -
- ]
510 [ + + + - : 88 : static const std::string& ip_multicast_str = Utils::get_env("VLINK_DDS_MULTICAST_IP");
+ - + - -
- ]
511 [ + + + - : 88 : static const std::string& peer_str = Utils::get_env("VLINK_DDS_PEER");
+ - + - -
- ]
512 [ + + + - : 88 : static const std::string& buf_str = Utils::get_env("VLINK_DDS_BUF");
+ - + - -
- ]
513 [ + + + - : 88 : static const std::string& mtu_str = Utils::get_env("VLINK_DDS_MTU");
+ - + - -
- ]
514 : :
515 [ + + + - : 88 : static bool enable_udp = Helpers::to_int(Utils::get_env("VLINK_DDS_UDP"), 1) != 0;
+ - + - -
- ]
516 [ + + + - : 88 : static bool enable_tcp = Helpers::to_int(Utils::get_env("VLINK_DDS_TCP"), 0) != 0;
+ - + - -
- ]
517 [ + + + - : 88 : static bool enable_shm = Helpers::to_int(Utils::get_env("VLINK_DDS_SHM"), 0) != 0;
+ - + - -
- ]
518 : :
519 [ + + + - : 88 : static bool enable_less_memory = Helpers::to_int(Utils::get_env("VLINK_DDS_LESS_MEMORY"), 0) != 0;
+ - + - -
- ]
520 : :
521 [ + + + - : 88 : static bool enable_ip_filter = Helpers::to_int(Utils::get_env("VLINK_DDS_IP_FILTER"), 0) != 0;
+ - + - -
- ]
522 : :
523 [ + + + - ]: 88 : static std::vector<std::string> default_ip_list = Utils::get_dds_default_address(enable_ip_filter);
524 : :
525 [ + - ]: 88 : std::string prop_ip_str = ip_str;
526 [ + - ]: 88 : std::string prop_ip_multicast_str = ip_multicast_str;
527 [ + - ]: 88 : std::string prop_peer_str = peer_str;
528 : 88 : size_t prop_buf = 0;
529 : 88 : size_t prop_mtu = 0;
530 : 88 : bool prop_enable_udp = enable_udp;
531 : 88 : bool prop_enable_tcp = enable_tcp;
532 : 88 : [[maybe_unused]] bool prop_enable_shm = enable_shm;
533 : 88 : [[maybe_unused]] bool prop_enable_less_memory = enable_less_memory;
534 : :
535 [ - + ]: 88 : if (!buf_str.empty()) {
536 [ # # ]: 0 : std::from_chars(buf_str.data(), buf_str.data() + buf_str.size(), prop_buf);
537 : : }
538 : :
539 [ - + ]: 88 : if (!mtu_str.empty()) {
540 [ # # ]: 0 : std::from_chars(mtu_str.data(), mtu_str.data() + mtu_str.size(), prop_mtu);
541 : : }
542 : :
543 [ + + ]: 108 : for (const auto& [prop, value] : properties) {
544 [ + + ]: 20 : if (!Helpers::has_startwith(prop, "dds.")) {
545 : 6 : continue;
546 : : }
547 : :
548 [ + + ]: 14 : if (prop == "dds.ip") {
549 [ + - ]: 2 : prop_ip_str = value;
550 [ + + ]: 12 : } else if (prop == "dds.multicast.ip") {
551 [ + - ]: 1 : prop_ip_multicast_str = value;
552 [ + + ]: 11 : } else if (prop == "dds.peer") {
553 [ + - ]: 1 : prop_peer_str = value;
554 [ + + ]: 10 : } else if (prop == "dds.buf") {
555 [ + - ]: 1 : std::from_chars(value.data(), value.data() + value.size(), prop_buf);
556 [ + + ]: 9 : } else if (prop == "dds.mtu") {
557 [ + - ]: 1 : std::from_chars(value.data(), value.data() + value.size(), prop_mtu);
558 [ + + ]: 8 : } else if (prop == "dds.udp") {
559 : 2 : prop_enable_udp = (value == "1");
560 [ + + ]: 6 : } else if (prop == "dds.tcp") {
561 : 2 : prop_enable_tcp = (value == "1");
562 [ + + ]: 4 : } else if (prop == "dds.shm") {
563 : 2 : prop_enable_shm = (value == "1");
564 [ + + ]: 2 : } else if (prop == "dds.less_memory") {
565 : 1 : prop_enable_less_memory = (value == "1");
566 : : } else {
567 [ + - ]: 1 : dds_qset_prop(dds_qos, prop.c_str(), value.c_str());
568 : : }
569 : : }
570 : :
571 : : (void)prop_enable_less_memory;
572 : :
573 [ + - + + ]: 88 : if (factory.domain_map_.find(domain_id) != factory.domain_map_.end()) {
574 : 4 : return;
575 : : }
576 : :
577 [ + - ]: 84 : ddsi_config config;
578 [ + - ]: 84 : ddsi_config_init_default(&config);
579 : :
580 [ + - ]: 84 : auto [domain_iter, inserted] = factory.domain_map_.try_emplace(domain_id);
581 : :
582 [ - + ]: 84 : if VUNLIKELY (!inserted) {
583 : 0 : return;
584 : : }
585 : :
586 : 84 : auto& domain_config = domain_iter->second;
587 : :
588 [ + + ]: 84 : if (prop_enable_udp) {
589 : 83 : config.transport_selector = DDSI_TRANS_UDP;
590 : : }
591 : :
592 : 84 : auto ssl_cfg = SslOptions::parse_from(properties);
593 : :
594 : 84 : bool ssl_cfg_valid = ssl_cfg.is_valid();
595 : :
596 [ + + + - ]: 84 : if (ssl_cfg_valid && !prop_enable_tcp) {
597 : 1 : prop_enable_tcp = true;
598 : : }
599 : :
600 [ + + ]: 84 : if (prop_enable_tcp) {
601 : 1 : config.transport_selector = DDSI_TRANS_TCP;
602 : 1 : config.tcp_port = 0;
603 : 1 : config.tcp_use_peeraddr_for_unicast = 1;
604 : 1 : config.compat_tcp_enable = DDSI_BOOLDEF_TRUE;
605 : : }
606 : :
607 : : #ifdef DDS_HAS_SSL
608 : :
609 : : if (ssl_cfg_valid && prop_enable_tcp) {
610 : : config.ssl_enable = 1;
611 : :
612 : : if (!ssl_cfg.cert_file.empty()) {
613 : : domain_config.ssl_keystore = ssl_cfg.cert_file;
614 : : } else if (!ssl_cfg.key_file.empty()) {
615 : : domain_config.ssl_keystore = ssl_cfg.key_file;
616 : : } else if (!ssl_cfg.ca_file.empty()) {
617 : : domain_config.ssl_keystore = ssl_cfg.ca_file;
618 : : }
619 : :
620 : : if (!domain_config.ssl_keystore.empty()) {
621 : : config.ssl_keystore = const_cast<char*>(domain_config.ssl_keystore.c_str());
622 : : }
623 : :
624 : : if (!ssl_cfg.key_password.empty()) {
625 : : domain_config.ssl_key_pass = ssl_cfg.key_password;
626 : : config.ssl_key_pass = const_cast<char*>(domain_config.ssl_key_pass.c_str());
627 : : }
628 : :
629 : : int provided =
630 : : (ssl_cfg.cert_file.empty() ? 0 : 1) + (ssl_cfg.key_file.empty() ? 0 : 1) + (ssl_cfg.ca_file.empty() ? 0 : 1);
631 : :
632 : : if VUNLIKELY (provided > 1) {
633 : : VLOG_W(
634 : : "DdscFactory: CycloneDDS only supports a single ssl_keystore (PEM/PKCS#12 with private key, certificate "
635 : : "and CA chain combined); ssl.cert/ssl.key/ssl.ca cannot be specified separately. Picked one and ignored "
636 : : "the others.");
637 : : }
638 : :
639 : : config.ssl_verify = ssl_cfg.verify_peer ? 1 : 0;
640 : : config.ssl_self_signed = ssl_cfg.verify_peer ? 0 : 1;
641 : :
642 : : if (!ssl_cfg.ciphers.empty()) {
643 : : domain_config.ssl_ciphers = ssl_cfg.ciphers;
644 : : config.ssl_ciphers = const_cast<char*>(domain_config.ssl_ciphers.c_str());
645 : : }
646 : : }
647 : : #else
648 : :
649 [ + + ]: 84 : if (ssl_cfg_valid) {
650 [ + - + - ]: 2 : VLOG_W("DdscFactory: ssl.* properties are set but CycloneDDS was built without DDS_HAS_SSL support.");
651 : : }
652 : : #endif
653 : :
654 [ + + ]: 84 : if (prop_buf > 0) {
655 : 1 : config.socket_sndbuf_size.min.isdefault = 1;
656 : 1 : config.socket_sndbuf_size.max.isdefault = 0;
657 : 1 : config.socket_sndbuf_size.max.value = static_cast<uint32_t>(prop_buf);
658 : :
659 : 1 : config.socket_rcvbuf_size.min.isdefault = 1;
660 : 1 : config.socket_rcvbuf_size.max.isdefault = 0;
661 : 1 : config.socket_rcvbuf_size.max.value = static_cast<uint32_t>(prop_buf);
662 : : }
663 : :
664 [ + + ]: 84 : if (prop_mtu > 0) {
665 [ + - ]: 1 : if (prop_mtu < config.fragment_size) {
666 : 1 : config.fragment_size = static_cast<uint32_t>(prop_mtu);
667 : : }
668 : :
669 [ + - ]: 1 : if (prop_mtu < config.max_rexmit_msg_size) {
670 : 1 : config.max_rexmit_msg_size = static_cast<uint32_t>(prop_mtu);
671 : : }
672 : :
673 : 1 : config.max_msg_size = static_cast<uint32_t>(prop_mtu);
674 : : }
675 : :
676 : : #ifdef DDS_HAS_SHM
677 : :
678 : : if (prop_enable_shm) {
679 : : config.enable_shm = 1;
680 : : } else {
681 : : config.enable_shm = 0;
682 : : }
683 : : #endif
684 : :
685 [ + + ]: 84 : if (prop_ip_multicast_str.empty()) {
686 : 83 : config.allowMulticast = DDSI_AMC_SPDP;
687 : : } else {
688 : 1 : config.allowMulticast = DDSI_AMC_TRUE;
689 : : }
690 : :
691 [ - + ]: 84 : if (prop_ip_str.empty()) {
692 [ # # ]: 0 : domain_config.network_interface_list = default_ip_list;
693 : : } else {
694 : 84 : domain_config.network_interface_list = Helpers::split_any(prop_ip_str);
695 : : }
696 : :
697 [ + - ]: 84 : if (!domain_config.network_interface_list.empty()) {
698 : : domain_config.network_interface_elements =
699 [ + - ]: 84 : std::make_unique<ddsi_config_network_interface_listelem[]>(domain_config.network_interface_list.size());
700 : :
701 [ + + ]: 168 : for (size_t i = 0; i < domain_config.network_interface_list.size(); ++i) {
702 : 84 : auto* interfaces = domain_config.network_interface_elements.get();
703 : :
704 [ - + ]: 84 : if (i < domain_config.network_interface_list.size() - 1) {
705 : 0 : interfaces[i].next = &(interfaces[i + 1]);
706 : : } else {
707 : 84 : interfaces[i].next = nullptr;
708 : : }
709 : :
710 : 84 : interfaces[i].cfg.automatic = 0;
711 : 84 : interfaces[i].cfg.name = nullptr;
712 [ + - ]: 84 : interfaces[i].cfg.address = const_cast<char*>(domain_config.network_interface_list.at(i).c_str());
713 : 84 : interfaces[i].cfg.prefer_multicast = 0;
714 : 84 : interfaces[i].cfg.presence_required = 1;
715 : 84 : interfaces[i].cfg.priority.isdefault = 1;
716 : 84 : interfaces[i].cfg.multicast = DDSI_BOOLDEF_TRUE;
717 : : }
718 : :
719 : 84 : config.network_interfaces = domain_config.network_interface_elements.get();
720 : : }
721 : :
722 : 84 : domain_config.peer_list = Helpers::split_any(prop_peer_str);
723 : :
724 [ + + ]: 84 : if (!domain_config.peer_list.empty()) {
725 [ + - ]: 1 : domain_config.peer_elements = std::make_unique<ddsi_config_peer_listelem[]>(domain_config.peer_list.size());
726 : :
727 [ + + ]: 2 : for (size_t i = 0; i < domain_config.peer_list.size(); ++i) {
728 : 1 : auto* peers = domain_config.peer_elements.get();
729 : :
730 [ - + ]: 1 : if (i < domain_config.peer_list.size() - 1) {
731 : 0 : peers[i].next = &(peers[i + 1]);
732 : : } else {
733 : 1 : peers[i].next = nullptr;
734 : : }
735 : :
736 [ + - ]: 1 : peers[i].peer = const_cast<char*>(domain_config.peer_list.at(i).c_str());
737 : : }
738 : :
739 : 1 : config.peers = domain_config.peer_elements.get();
740 : : }
741 : :
742 [ + - ]: 84 : auto domain = dds_create_domain_with_rawconfig(domain_id, &config);
743 : :
744 [ - + ]: 84 : if VUNLIKELY (domain <= 0) {
745 [ # # ]: 0 : factory.domain_map_.erase(domain_iter);
746 : 0 : return;
747 : : }
748 : :
749 : 84 : domain_config.entity = domain;
750 [ + - + + : 96 : }
+ + + + ]
751 : :
752 : : } // namespace vlink
|
