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 "./dds_factory.h"
25 : :
26 : : #include <charconv>
27 : : #include <limits>
28 : : #include <memory>
29 : : #include <string>
30 : : #include <utility>
31 : : #include <vector>
32 : :
33 : : #include "./base/helpers.h"
34 : : #include "./base/utils.h"
35 : : #include "./dds_qos.h"
36 : : #include "./dds_topic.h"
37 : : #include "./extension/qos_profile.h"
38 : : #include "./impl/ssl_options.h"
39 : :
40 : : namespace vlink {
41 : :
42 : : // DdsFactory
43 [ + - ]: 13 : DdsFactory::DdsFactory() {
44 : 13 : Bytes::init_memory_pool();
45 : :
46 [ - + ]: 13 : if VUNLIKELY (DdsConf::get_thread_count() != 1) {
47 [ # # # # ]: 0 : VLOG_W("DdsFactory: Dds does not support setting thread count.");
48 : : }
49 : :
50 [ + + ]: 221 : for (const auto& [name, qos] : QosProfile::get_available_qos_map()) {
51 [ + - ]: 208 : DdsConf::register_qos_internal(name, qos);
52 : : }
53 : :
54 [ + - + - ]: 26 : std::string dds_debug_str = Utils::get_env("VLINK_DDS_DEBUG");
55 : :
56 [ - + ]: 13 : if (dds_debug_str == "1") {
57 [ # # ]: 0 : dds::Log::SetVerbosity(dds::Log::Kind::Info);
58 : : } else {
59 [ + - ]: 13 : dds::Log::SetVerbosity(dds::Log::Kind::Error);
60 : : }
61 : :
62 [ + - ]: 13 : dds_factory_ = dds::DomainParticipantFactory::get_instance();
63 : :
64 [ + - + - ]: 26 : std::string qos_file = Utils::get_env("VLINK_FASTDDS_QOS_FILE");
65 : :
66 [ - + ]: 13 : if (!qos_file.empty()) {
67 [ # # ]: 0 : dds_factory_->load_XML_profiles_file(qos_file);
68 : : }
69 : :
70 [ + - + - ]: 26 : std::string default_event_qos_str = Utils::get_env("VLINK_DDS_EVENT_QOS");
71 [ + - + - ]: 26 : std::string default_method_qos_str = Utils::get_env("VLINK_DDS_METHOD_QOS");
72 [ + - + - ]: 26 : std::string default_field_qos_str = Utils::get_env("VLINK_DDS_FIELD_QOS");
73 : :
74 [ + - ]: 13 : if (default_event_qos_str.empty()) {
75 : 13 : default_event_qos_ = QosProfile::kEvent;
76 : : } else {
77 [ # # ]: 0 : default_event_qos_ = DdsConf::find_qos(default_event_qos_str);
78 : : }
79 : :
80 [ + - ]: 13 : if (default_method_qos_str.empty()) {
81 : 13 : default_method_qos_ = QosProfile::kMethod;
82 : : } else {
83 [ # # ]: 0 : default_method_qos_ = DdsConf::find_qos(default_method_qos_str);
84 : : }
85 : :
86 [ + - ]: 13 : if (default_field_qos_str.empty()) {
87 : 13 : default_field_qos_ = QosProfile::kField;
88 : : } else {
89 [ # # ]: 0 : default_field_qos_ = DdsConf::find_qos(default_field_qos_str);
90 : : }
91 : 13 : }
92 : :
93 : 13 : DdsFactory::~DdsFactory() = default;
94 : :
95 : 5 : std::vector<std::tuple<std::string, std::string>> DdsFactory::get_discovered_topics(int32_t _domain) {
96 : 5 : std::vector<std::tuple<std::string, std::string>> topics;
97 : :
98 [ + + + - : 5 : static auto& factory = DdsFactory::get();
+ - - - ]
99 [ + - ]: 5 : std::lock_guard lifecycle_lock(factory.participant_mtx_);
100 : :
101 [ + - ]: 5 : auto* part = factory.dds_factory_->lookup_participant(_domain);
102 : :
103 [ + + ]: 5 : if VUNLIKELY (!part) {
104 : 4 : return topics;
105 : : }
106 : :
107 : 1 : std::vector<dds::InstanceHandle_t> topic_handles;
108 [ + - ]: 1 : part->get_discovered_topics(topic_handles);
109 : :
110 [ + - ]: 1 : topics.reserve(topic_handles.size());
111 : :
112 [ - + ]: 1 : for (const auto& instance : topic_handles) {
113 [ # # ]: 0 : dds::builtin::TopicBuiltinTopicData topic_data;
114 [ # # ]: 0 : part->get_discovered_topic_data(topic_data, instance);
115 [ # # ]: 0 : topics.emplace_back(std::forward_as_tuple(std::move(topic_data.name), std::move(topic_data.type_name)));
116 : 0 : }
117 : :
118 : 1 : return topics;
119 : 5 : }
120 : :
121 : 2 : bool DdsFactory::load_global_qos_file(const std::string& filepath) {
122 [ + - + - : 2 : static auto& factory = DdsFactory::get();
+ - - - ]
123 : :
124 : : #ifdef VLINK_SUPPORT_DDS_V3
125 : : return factory.dds_factory_->load_XML_profiles_file(filepath) == dds::RETCODE_OK;
126 : : #else
127 [ + - ]: 2 : return factory.dds_factory_->load_XML_profiles_file(filepath) == ReturnCode_t::RETCODE_OK;
128 : : #endif
129 : : }
130 : :
131 : 256 : std::shared_ptr<dds::DomainParticipant> DdsFactory::create_participant(uint8_t type, const DdsConf& conf,
132 : : const Conf::PropertiesMap& properties) {
133 [ + + + - : 256 : static auto& factory = DdsFactory::get();
+ - - - ]
134 : :
135 [ + - + - ]: 256 : const auto& dds_qos_ext = get_qos_ext(conf.qos_ext, "part");
136 [ + - ]: 256 : const auto& id = std::make_tuple(type, conf.domain, dds_qos_ext, properties);
137 : :
138 [ + - ]: 256 : std::lock_guard lifecycle_lock(factory.participant_mtx_);
139 [ + - ]: 256 : std::unique_lock lock(factory.mtx_);
140 : :
141 [ + - ]: 256 : std::shared_ptr<dds::DomainParticipant> part = get_weak_ptr(factory.part_map_, id).lock();
142 : :
143 [ + + ]: 256 : if (!part) {
144 [ + - ]: 130 : lock.unlock();
145 : :
146 : 130 : dds::DomainParticipant* ptr = nullptr;
147 : :
148 [ + + ]: 130 : if (dds_qos_ext.empty()) {
149 [ + - ]: 129 : auto dds_qos = dds::PARTICIPANT_QOS_DEFAULT;
150 : :
151 [ + - ]: 129 : set_participant_qos(dds_qos, properties);
152 : :
153 [ + - ]: 129 : ptr = factory.dds_factory_->create_participant(conf.domain, dds_qos, nullptr, dds::StatusMask::all());
154 : 129 : } else {
155 [ + - ]: 1 : ptr = factory.dds_factory_->create_participant_with_profile(conf.domain, dds_qos_ext, nullptr,
156 : 2 : dds::StatusMask::all());
157 : : }
158 : :
159 [ + + ]: 130 : if VUNLIKELY (!ptr) {
160 [ + - + - ]: 4 : VLOG_E("DdsFactory: Failed to create participant.");
161 : 2 : return nullptr;
162 : : }
163 : :
164 [ + - + - ]: 256 : part = std::shared_ptr<dds::DomainParticipant>(ptr, [id](dds::DomainParticipant* part) {
165 [ + - ]: 128 : std::lock_guard lifecycle_lock(factory.participant_mtx_);
166 : :
167 : : {
168 [ + - ]: 128 : std::lock_guard lock(factory.mtx_);
169 [ + - ]: 128 : auto iter = factory.part_map_.find(id);
170 : :
171 [ + - + - : 128 : if (iter != factory.part_map_.end() && iter->second.expired()) {
+ - ]
172 [ + - ]: 128 : factory.part_map_.erase(iter);
173 : : }
174 : 128 : }
175 : :
176 [ + - ]: 128 : factory.dds_factory_->delete_participant(part);
177 : 256 : });
178 : :
179 [ + - ]: 128 : lock.lock();
180 : :
181 [ + - ]: 128 : auto [iter, inserted] = factory.part_map_.emplace(id, part);
182 : :
183 [ - + ]: 128 : if (!inserted) {
184 : 0 : auto inserted_part = iter->second.lock();
185 : :
186 [ # # ]: 0 : if VLIKELY (inserted_part) {
187 [ # # ]: 0 : lock.unlock();
188 : 0 : part = std::move(inserted_part);
189 : : } else {
190 : 0 : iter->second = part;
191 : : }
192 : 0 : }
193 : : }
194 : :
195 : 254 : return part;
196 : 256 : }
197 : :
198 : 303 : std::shared_ptr<dds::Topic> DdsFactory::create_topic(uint8_t type, const DdsConf& conf, dds::DomainParticipant* part,
199 : : bool is_cdr_type, std::string topic, std::string cdr_type_name) {
200 [ + + + - : 303 : static auto& factory = DdsFactory::get();
+ - - - ]
201 : :
202 : 303 : dds::TypeSupport type_support;
203 : 303 : dds::TypeSupport native_type;
204 : :
205 [ + + ]: 303 : if (topic.empty()) {
206 [ + - ]: 209 : topic = conf.topic;
207 : : }
208 : :
209 [ + - ]: 303 : Function<void*()> type_support_callback = DdsConf::find_type_support(topic);
210 : :
211 [ + + ]: 303 : if (is_cdr_type) {
212 [ - + ]: 2 : if (type_support_callback) {
213 [ # # # # ]: 0 : native_type.reset(static_cast<dds::TopicDataType*>(type_support_callback()));
214 : :
215 [ # # ]: 0 : if VUNLIKELY (!native_type) {
216 [ # # # # ]: 0 : VLOG_F("DdsFactory: Topic ", topic, " CDR typesupport creation failed.");
217 : : }
218 : :
219 : 0 : const auto& native_type_name = native_type.get_type_name();
220 : :
221 [ # # # # : 0 : if VUNLIKELY (!cdr_type_name.empty() && cdr_type_name != native_type_name) {
# # ]
222 [ # # # # ]: 0 : CLOG_W("DdsFactory: Topic [%s] CDR type [%s] => [%s].", topic.c_str(), cdr_type_name.c_str(),
223 : : native_type_name.c_str());
224 : : }
225 : :
226 [ # # ]: 0 : cdr_type_name = native_type_name;
227 : : }
228 : :
229 [ - + ]: 2 : if VUNLIKELY (cdr_type_name.empty()) {
230 [ # # # # ]: 0 : VLOG_F("DdsFactory: Topic ", topic, " has no CDR type name.");
231 : : }
232 : : } else {
233 [ - + ]: 301 : if VUNLIKELY (type_support_callback) {
234 [ # # # # ]: 0 : VLOG_F("DdsFactory: Topic ", topic, " does not support BuiltIn::Raw.");
235 : : }
236 : :
237 [ + - ]: 301 : std::lock_guard raw_lock(factory.raw_typesupport_mtx_);
238 : :
239 [ + + ]: 301 : if VUNLIKELY (!factory.raw_typesupport_) {
240 [ + - + - : 12 : factory.raw_typesupport_.reset(new BuiltInRawPubSubType); // NOLINT(modernize-make-shared)
+ - ]
241 : : }
242 : 301 : }
243 : :
244 [ + - + - ]: 303 : const auto& dds_qos_ext = get_qos_ext(conf.qos_ext, "topic");
245 : :
246 [ + + ]: 303 : if VUNLIKELY (!part) {
247 [ + - + - ]: 4 : VLOG_E("DdsFactory: Cannot create topic without participant.");
248 : 2 : return nullptr;
249 : : }
250 : :
251 [ + - ]: 301 : const auto& id = std::make_tuple(type, conf.domain, topic, part);
252 : :
253 [ + - ]: 301 : std::unique_lock lock(factory.mtx_);
254 [ + - ]: 301 : std::shared_ptr<dds::Topic> dds_topic = get_weak_ptr(factory.topic_map_, id).lock();
255 : :
256 [ + + ]: 301 : if (!dds_topic) {
257 [ + - ]: 170 : lock.unlock();
258 : :
259 [ + + ]: 170 : if (is_cdr_type) {
260 [ + - + - : 2 : type_support.reset(new DdsCdrPubSubType(cdr_type_name, std::move(native_type)));
+ - ]
261 : : } else {
262 [ + - ]: 168 : std::lock_guard raw_lock(factory.raw_typesupport_mtx_);
263 : 168 : type_support = factory.raw_typesupport_;
264 : 168 : }
265 : :
266 [ + - ]: 170 : if VLIKELY (type_support) {
267 [ + - ]: 170 : part->register_type(type_support);
268 : : } else {
269 [ # # # # ]: 0 : VLOG_F("DdsFactory: Topic ", topic, " registration failed.");
270 : : }
271 : :
272 : 170 : dds::Topic* ptr = nullptr;
273 : :
274 [ + + ]: 170 : if (dds_qos_ext.empty()) {
275 [ + - ]: 169 : auto dds_qos = dds::TOPIC_QOS_DEFAULT;
276 : :
277 [ + - ]: 169 : ptr = part->create_topic(topic, type_support.get_type_name(), dds_qos);
278 : 0 : } else {
279 [ + - ]: 1 : ptr = part->create_topic_with_profile(topic, type_support.get_type_name(), dds_qos_ext);
280 : : }
281 : :
282 [ + + ]: 170 : if VUNLIKELY (!ptr) {
283 [ + - + - ]: 2 : VLOG_E("DdsFactory: Failed to create topic: ", topic, ".");
284 : 1 : return nullptr;
285 : : }
286 : :
287 [ + - + - ]: 338 : dds_topic = std::shared_ptr<dds::Topic>(ptr, [id](dds::Topic* topic) {
288 : : {
289 [ + - ]: 169 : std::lock_guard lock(factory.mtx_);
290 [ + - ]: 169 : auto iter = factory.topic_map_.find(id);
291 : :
292 [ + - + - : 169 : if (iter != factory.topic_map_.end() && iter->second.expired()) {
+ - ]
293 [ + - ]: 169 : factory.topic_map_.erase(iter);
294 : : }
295 : 169 : }
296 : :
297 : 169 : auto* participant = const_cast<dds::DomainParticipant*>(topic->get_participant());
298 : 169 : participant->delete_topic(topic);
299 : 338 : });
300 : :
301 [ + - ]: 169 : lock.lock();
302 : :
303 [ + - ]: 169 : auto [iter, inserted] = factory.topic_map_.emplace(id, dds_topic);
304 : :
305 [ - + ]: 169 : if (!inserted) {
306 : 0 : auto inserted_topic = iter->second.lock();
307 : :
308 [ # # ]: 0 : if VLIKELY (inserted_topic) {
309 [ # # ]: 0 : lock.unlock();
310 : 0 : dds_topic = std::move(inserted_topic);
311 : : } else {
312 : 0 : iter->second = dds_topic;
313 : : }
314 : 0 : }
315 : : } else {
316 [ - + ]: 131 : const auto& expected_type_name = is_cdr_type ? cdr_type_name : factory.raw_typesupport_.get_type_name();
317 : :
318 [ - + ]: 131 : if VUNLIKELY (dds_topic->get_type_name() != expected_type_name) {
319 [ # # # # ]: 0 : VLOG_F("DdsFactory: Topic ", topic, " type mismatch.");
320 : : }
321 : :
322 [ + - ]: 131 : type_support = part->find_type(dds_topic->get_type_name());
323 : :
324 [ - + ]: 131 : if (!type_support) {
325 [ # # ]: 0 : if (is_cdr_type) {
326 [ # # # # : 0 : type_support.reset(new DdsCdrPubSubType(cdr_type_name, std::move(native_type)));
# # ]
327 : : } else {
328 [ # # ]: 0 : std::lock_guard raw_lock(factory.raw_typesupport_mtx_);
329 : 0 : type_support = factory.raw_typesupport_;
330 : 0 : }
331 : : }
332 : :
333 [ + - ]: 131 : if VLIKELY (type_support) {
334 [ + - ]: 131 : part->register_type(type_support);
335 : : } else {
336 [ # # # # ]: 0 : VLOG_F("DdsFactory: Topic ", topic, " registration failed.");
337 : : }
338 : : }
339 : :
340 [ + + ]: 300 : const auto& expected_type_name = is_cdr_type ? cdr_type_name : factory.raw_typesupport_.get_type_name();
341 : :
342 [ - + ]: 300 : if VUNLIKELY (dds_topic->get_type_name() != expected_type_name) {
343 [ # # # # ]: 0 : VLOG_F("DdsFactory: Topic ", topic, " type mismatch.");
344 : : }
345 : :
346 : 300 : return dds_topic;
347 : 303 : }
348 : :
349 : 47 : std::pair<std::shared_ptr<dds::Topic>, std::shared_ptr<dds::Topic>> DdsFactory::create_method_topic(
350 : : uint8_t type, const DdsConf& conf, dds::DomainParticipant* part, bool is_cdr_type,
351 : : const std::string& cdr_type_names) {
352 [ + - ]: 47 : const std::string& resp_topic = conf.topic + DdsConf::kRespSuffix;
353 : :
354 [ + - - + : 47 : if VUNLIKELY (conf.topic.empty() || resp_topic.empty()) {
- + ]
355 [ # # # # ]: 0 : VLOG_F("DdsFactory: Method conf topic error.");
356 : : }
357 : :
358 [ - + ]: 47 : if VUNLIKELY (conf.topic == resp_topic) {
359 [ # # # # ]: 0 : VLOG_F("DdsFactory: Method conf topic req and resp cannot be equal.");
360 : : }
361 : :
362 : 47 : std::string req_cdr_type_name;
363 : 47 : std::string resp_cdr_type_name;
364 : :
365 [ - + ]: 47 : if (is_cdr_type) {
366 : 0 : const auto separator = cdr_type_names.find('|');
367 : :
368 [ # # ]: 0 : if (separator == std::string::npos) {
369 [ # # ]: 0 : req_cdr_type_name = cdr_type_names;
370 [ # # ]: 0 : resp_cdr_type_name = cdr_type_names;
371 : : } else {
372 [ # # ]: 0 : if VUNLIKELY (separator == 0U) {
373 [ # # # # ]: 0 : VLOG_F("DdsFactory: CDR method request type name is empty.");
374 : : }
375 : :
376 [ # # ]: 0 : if VUNLIKELY (separator + 1U == cdr_type_names.size()) {
377 [ # # # # ]: 0 : VLOG_F("DdsFactory: CDR method response type name is empty.");
378 : : }
379 : :
380 [ # # ]: 0 : req_cdr_type_name = cdr_type_names.substr(0U, separator);
381 [ # # ]: 0 : resp_cdr_type_name = cdr_type_names.substr(separator + 1U);
382 : : }
383 : : }
384 : :
385 [ + - + - ]: 94 : return {create_topic(type, conf, part, is_cdr_type, conf.topic, std::move(req_cdr_type_name)),
386 [ + - + - ]: 94 : create_topic(type, conf, part, is_cdr_type, resp_topic, std::move(resp_cdr_type_name))};
387 : 47 : }
388 : :
389 : 157 : std::shared_ptr<dds::Publisher> DdsFactory::create_publisher(uint8_t type, const DdsConf& conf,
390 : : dds::DomainParticipant* part) {
391 [ + + + - : 157 : static auto& factory = DdsFactory::get();
+ - - - ]
392 : :
393 [ + - + - ]: 157 : const auto& dds_qos_ext = get_qos_ext(conf.qos_ext, "pub");
394 [ + - + - ]: 157 : const auto& writer_qos = get_qos_ext(conf.qos_ext, "writer");
395 : :
396 [ + + ]: 157 : if VUNLIKELY (!part) {
397 [ + - + - ]: 4 : VLOG_E("DdsFactory: Cannot create publisher without participant.");
398 : 2 : return nullptr;
399 : : }
400 : :
401 [ + - ]: 155 : const auto& id = std::make_tuple(type, conf.domain, conf.qos, dds_qos_ext, writer_qos, part);
402 : :
403 [ + - ]: 155 : std::unique_lock lock(factory.mtx_);
404 [ + - ]: 155 : std::shared_ptr<dds::Publisher> publisher = get_weak_ptr(factory.publisher_map_, id).lock();
405 : :
406 [ + + ]: 155 : if (!publisher) {
407 [ + - ]: 140 : lock.unlock();
408 : :
409 : 140 : dds::Publisher* ptr = nullptr;
410 : :
411 [ + + ]: 140 : if (dds_qos_ext.empty()) {
412 [ + - ]: 139 : auto dds_qos = dds::PUBLISHER_QOS_DEFAULT;
413 : :
414 [ + - ]: 139 : ptr = part->create_publisher(dds_qos, nullptr);
415 : 139 : } else {
416 [ + - ]: 1 : ptr = part->create_publisher_with_profile(dds_qos_ext, nullptr);
417 : : }
418 : :
419 [ + + ]: 140 : if VUNLIKELY (!ptr) {
420 [ + - + - ]: 2 : VLOG_E("DdsFactory: Failed to create publisher.");
421 : 1 : return nullptr;
422 : : }
423 : :
424 [ + - + - ]: 278 : publisher = std::shared_ptr<dds::Publisher>(ptr, [id](dds::Publisher* publisher) {
425 : : {
426 [ + - ]: 139 : std::lock_guard lock(factory.mtx_);
427 [ + - ]: 139 : auto iter = factory.publisher_map_.find(id);
428 : :
429 [ + - + - : 139 : if (iter != factory.publisher_map_.end() && iter->second.expired()) {
+ - ]
430 [ + - ]: 139 : factory.publisher_map_.erase(iter);
431 : : }
432 : 139 : }
433 : :
434 : 139 : auto* participant = const_cast<dds::DomainParticipant*>(publisher->get_participant());
435 : 139 : participant->delete_publisher(publisher);
436 : 278 : });
437 : :
438 [ + - ]: 139 : lock.lock();
439 : :
440 [ + - ]: 139 : auto [iter, inserted] = factory.publisher_map_.emplace(id, publisher);
441 : :
442 [ - + ]: 139 : if (!inserted) {
443 : 0 : auto inserted_publisher = iter->second.lock();
444 : :
445 [ # # ]: 0 : if VLIKELY (inserted_publisher) {
446 [ # # ]: 0 : lock.unlock();
447 : 0 : publisher = std::move(inserted_publisher);
448 : : } else {
449 : 0 : iter->second = publisher;
450 : : }
451 : 0 : }
452 : : }
453 : :
454 : 154 : return publisher;
455 : 157 : }
456 : :
457 : 146 : std::shared_ptr<dds::Subscriber> DdsFactory::create_subscriber(uint8_t type, const DdsConf& conf,
458 : : dds::DomainParticipant* part) {
459 [ + + + - : 146 : static auto& factory = DdsFactory::get();
+ - - - ]
460 : :
461 [ + - + - ]: 146 : const auto& dds_qos_ext = get_qos_ext(conf.qos_ext, "sub");
462 [ + - + - ]: 146 : const auto& reader_qos = get_qos_ext(conf.qos_ext, "reader");
463 : :
464 [ - + ]: 146 : if VUNLIKELY (!part) {
465 [ # # # # ]: 0 : VLOG_E("DdsFactory: Cannot create subscriber without participant.");
466 : 0 : return nullptr;
467 : : }
468 : :
469 [ + - ]: 146 : const auto& id = std::make_tuple(type, conf.domain, conf.qos, dds_qos_ext, reader_qos, part);
470 : :
471 [ + - ]: 146 : std::unique_lock lock(factory.mtx_);
472 : :
473 [ + - ]: 146 : std::shared_ptr<dds::Subscriber> subscriber = get_weak_ptr(factory.subscriber_map_, id).lock();
474 : :
475 [ + + ]: 146 : if (!subscriber) {
476 [ + - ]: 134 : lock.unlock();
477 : 134 : dds::Subscriber* ptr = nullptr;
478 : :
479 [ + + ]: 134 : if (dds_qos_ext.empty()) {
480 [ + - ]: 133 : auto dds_qos = dds::SUBSCRIBER_QOS_DEFAULT;
481 [ + - ]: 133 : ptr = part->create_subscriber(dds_qos, nullptr);
482 : 133 : } else {
483 [ + - ]: 1 : ptr = part->create_subscriber_with_profile(dds_qos_ext, nullptr);
484 : : }
485 : :
486 [ + + ]: 134 : if VUNLIKELY (!ptr) {
487 [ + - + - ]: 2 : VLOG_E("DdsFactory: Failed to create subscriber.");
488 : 1 : return nullptr;
489 : : }
490 : :
491 [ + - + - ]: 266 : subscriber = std::shared_ptr<dds::Subscriber>(ptr, [id](dds::Subscriber* subscriber) {
492 : : {
493 [ + - ]: 133 : std::lock_guard lock(factory.mtx_);
494 [ + - ]: 133 : auto iter = factory.subscriber_map_.find(id);
495 : :
496 [ + - + - : 133 : if (iter != factory.subscriber_map_.end() && iter->second.expired()) {
+ - ]
497 [ + - ]: 133 : factory.subscriber_map_.erase(iter);
498 : : }
499 : 133 : }
500 : 133 : auto* participant = const_cast<dds::DomainParticipant*>(subscriber->get_participant());
501 : 133 : participant->delete_subscriber(subscriber);
502 : 266 : });
503 : :
504 [ + - ]: 133 : lock.lock();
505 : :
506 [ + - ]: 133 : auto [iter, inserted] = factory.subscriber_map_.emplace(id, subscriber);
507 : :
508 [ - + ]: 133 : if (!inserted) {
509 : 0 : auto inserted_subscriber = iter->second.lock();
510 : :
511 [ # # ]: 0 : if VLIKELY (inserted_subscriber) {
512 [ # # ]: 0 : lock.unlock();
513 : 0 : subscriber = std::move(inserted_subscriber);
514 : : } else {
515 : 0 : iter->second = subscriber;
516 : : }
517 : 0 : }
518 : : }
519 : :
520 : 145 : return subscriber;
521 : 146 : }
522 : :
523 : 154 : std::shared_ptr<dds::DataWriter> DdsFactory::create_datawriter(uint8_t type, const DdsConf& conf,
524 : : dds::Publisher* publisher, dds::Topic* topic,
525 : : dds::DataWriterListener* listener, bool is_cdr_type) {
526 [ + + + - : 154 : static auto& factory = DdsFactory::get();
+ - - - ]
527 : :
528 [ + - + - ]: 154 : const auto& dds_qos_ext = get_qos_ext(conf.qos_ext, "writer");
529 : :
530 [ + + - + : 154 : if VUNLIKELY (!publisher || !topic) {
+ + ]
531 [ + - + - ]: 2 : VLOG_E("DdsFactory: Cannot create datawriter without publisher/topic.");
532 : 1 : return nullptr;
533 : : }
534 : :
535 : 153 : dds::DataWriter* ptr = nullptr;
536 : :
537 [ + + ]: 153 : if (dds_qos_ext.empty()) {
538 [ + - ]: 152 : auto dds_qos = dds::DATAWRITER_QOS_DEFAULT;
539 : :
540 [ + + ]: 152 : if (conf.qos.empty()) {
541 [ + + - + ]: 129 : if ((type & kPublisher) || (type & kSubscriber)) {
542 [ + - ]: 54 : convert_qos(dds_qos, factory.default_event_qos_, conf.depth);
543 [ + + + + ]: 75 : } else if ((type & kClient) || (type & kServer)) {
544 [ + - ]: 53 : convert_qos(dds_qos, factory.default_method_qos_, conf.depth);
545 [ - + - - ]: 22 : } else if ((type & kSetter) || (type & kGetter)) {
546 [ + - ]: 22 : convert_qos(dds_qos, factory.default_field_qos_, conf.depth);
547 : : }
548 : : } else {
549 [ + - + - ]: 23 : convert_qos(dds_qos, DdsConf::find_qos(conf.qos), conf.depth);
550 : : }
551 : :
552 [ + + ]: 152 : if VUNLIKELY (is_cdr_type) {
553 : 1 : dds_qos.endpoint().history_memory_policy = rtps::PREALLOCATED_WITH_REALLOC_MEMORY_MODE;
554 : : }
555 : :
556 [ + - ]: 152 : ptr = publisher->create_datawriter(topic, dds_qos, listener);
557 [ - + ]: 153 : } else if VUNLIKELY (is_cdr_type) {
558 [ # # ]: 0 : auto dds_qos = dds::DATAWRITER_QOS_DEFAULT;
559 [ # # ]: 0 : const auto ret = publisher->get_datawriter_qos_from_profile(dds_qos_ext, dds_qos);
560 : :
561 : : #ifdef VLINK_SUPPORT_DDS_V3
562 : : if VLIKELY (ret == dds::RETCODE_OK) {
563 : : #else
564 [ # # ]: 0 : if VLIKELY (ret == ReturnCode_t::RETCODE_OK) {
565 : : #endif
566 : 0 : dds_qos.endpoint().history_memory_policy = rtps::PREALLOCATED_WITH_REALLOC_MEMORY_MODE;
567 [ # # ]: 0 : ptr = publisher->create_datawriter(topic, dds_qos, listener);
568 : : }
569 : 0 : } else {
570 [ + - ]: 1 : ptr = publisher->create_datawriter_with_profile(topic, dds_qos_ext, listener);
571 : : }
572 : :
573 [ + + ]: 153 : if VUNLIKELY (!ptr) {
574 [ + - + - ]: 8 : VLOG_E("DdsFactory: Failed to create datawriter.");
575 : 4 : return nullptr;
576 : : }
577 : :
578 : 149 : return std::shared_ptr<dds::DataWriter>(ptr, [](dds::DataWriter* writer) {
579 [ + - ]: 149 : writer->set_listener(nullptr, dds::StatusMask::none());
580 : 149 : auto* publisher = const_cast<dds::Publisher*>(writer->get_publisher());
581 : 149 : publisher->delete_datawriter(writer);
582 [ + - ]: 298 : });
583 : 154 : }
584 : :
585 : 135 : std::shared_ptr<dds::DataReader> DdsFactory::create_datareader(uint8_t type, const DdsConf& conf,
586 : : dds::Subscriber* subscriber, dds::Topic* topic,
587 : : dds::DataReaderListener* listener, bool is_cdr_type) {
588 [ + + + - : 135 : static auto& factory = DdsFactory::get();
+ - - - ]
589 : :
590 [ + - + - ]: 135 : const auto& dds_qos_ext = get_qos_ext(conf.qos_ext, "reader");
591 : :
592 [ + - - + : 135 : if VUNLIKELY (!subscriber || !topic) {
- + ]
593 [ # # # # ]: 0 : VLOG_E("DdsFactory: Cannot create datareader without subscriber/topic.");
594 : 0 : return nullptr;
595 : : }
596 : :
597 : 135 : dds::DataReader* ptr = nullptr;
598 : :
599 [ + - ]: 135 : if (dds_qos_ext.empty()) {
600 [ + - ]: 135 : auto dds_qos = dds::DATAREADER_QOS_DEFAULT;
601 : :
602 [ + + ]: 135 : if (conf.qos.empty()) {
603 [ + - + + ]: 114 : if ((type & kPublisher) || (type & kSubscriber)) {
604 [ + - ]: 41 : convert_qos(dds_qos, factory.default_event_qos_, conf.depth);
605 [ + + + + ]: 73 : } else if ((type & kClient) || (type & kServer)) {
606 [ + - ]: 51 : convert_qos(dds_qos, factory.default_method_qos_, conf.depth);
607 [ + - + - ]: 22 : } else if ((type & kSetter) || (type & kGetter)) {
608 [ + - ]: 22 : convert_qos(dds_qos, factory.default_field_qos_, conf.depth);
609 : : }
610 : : } else {
611 [ + - + - ]: 21 : convert_qos(dds_qos, DdsConf::find_qos(conf.qos), conf.depth);
612 : : }
613 : :
614 [ - + ]: 135 : if VUNLIKELY (is_cdr_type) {
615 : 0 : dds_qos.endpoint().history_memory_policy = rtps::PREALLOCATED_WITH_REALLOC_MEMORY_MODE;
616 : : }
617 : :
618 [ + - + - ]: 135 : ptr = subscriber->create_datareader(topic, dds_qos, listener);
619 [ - - ]: 135 : } else if VUNLIKELY (is_cdr_type) {
620 [ # # ]: 0 : auto dds_qos = dds::DATAREADER_QOS_DEFAULT;
621 [ # # ]: 0 : const auto ret = subscriber->get_datareader_qos_from_profile(dds_qos_ext, dds_qos);
622 : :
623 : : #ifdef VLINK_SUPPORT_DDS_V3
624 : : if VLIKELY (ret == dds::RETCODE_OK) {
625 : : #else
626 [ # # ]: 0 : if VLIKELY (ret == ReturnCode_t::RETCODE_OK) {
627 : : #endif
628 : 0 : dds_qos.endpoint().history_memory_policy = rtps::PREALLOCATED_WITH_REALLOC_MEMORY_MODE;
629 [ # # # # ]: 0 : ptr = subscriber->create_datareader(topic, dds_qos, listener);
630 : : }
631 : 0 : } else {
632 [ # # # # ]: 0 : ptr = subscriber->create_datareader_with_profile(topic, dds_qos_ext, listener);
633 : : }
634 : :
635 [ + + ]: 135 : if VUNLIKELY (!ptr) {
636 [ + - + - ]: 6 : VLOG_E("DdsFactory: Failed to create datareader.");
637 : 3 : return nullptr;
638 : : }
639 : :
640 : 132 : return std::shared_ptr<dds::DataReader>(ptr, [](dds::DataReader* reader) {
641 [ + - ]: 132 : reader->set_listener(nullptr, dds::StatusMask::none());
642 : 132 : auto* subscriber = const_cast<dds::Subscriber*>(reader->get_subscriber());
643 : 132 : subscriber->delete_datareader(reader);
644 [ + - ]: 264 : });
645 : 135 : }
646 : :
647 : 193 : bool DdsFactory::write_data(dds::DataWriter* writer, const Bytes& bytes, uint64_t id) {
648 [ - + ]: 193 : if VUNLIKELY (bytes.is_ptr()) {
649 [ # # # # ]: 0 : VLOG_E("DdsFactory: write_data() type mismatch, expected raw bytes but received ptr type.");
650 : 0 : return false;
651 : : }
652 : :
653 [ + - ]: 193 : BuiltInRaw raw;
654 [ + - ]: 193 : raw.id() = id;
655 : :
656 [ + - ]: 193 : raw.data().shallow_copy(bytes);
657 : :
658 : : #ifdef VLINK_SUPPORT_DDS_V3
659 : : return writer->write(&raw) == dds::RETCODE_OK;
660 : : #else
661 [ + - ]: 193 : return writer->write(&raw);
662 : : #endif
663 : 193 : }
664 : :
665 : 0 : bool DdsFactory::write_cdr_data(dds::DataWriter* writer, const Bytes& bytes, rtps::WriteParams* params) {
666 [ # # # # : 0 : if VUNLIKELY (bytes.is_ptr() || bytes.size() < 4U || bytes.size() > std::numeric_limits<uint32_t>::max()) {
# # # # #
# ]
667 [ # # # # ]: 0 : VLOG_E("DdsFactory: write_cdr_data() expected an encapsulated CDR payload.");
668 : 0 : return false;
669 : : }
670 : :
671 : : #ifdef VLINK_SUPPORT_DDS_V3
672 : :
673 : : if (params) {
674 : : return writer->write(const_cast<Bytes*>(&bytes), *params) == dds::RETCODE_OK;
675 : : }
676 : :
677 : : return writer->write(const_cast<Bytes*>(&bytes)) == dds::RETCODE_OK;
678 : : #else
679 : :
680 [ # # ]: 0 : if (params) {
681 : 0 : return writer->write(const_cast<Bytes*>(&bytes), *params);
682 : : }
683 : :
684 : 0 : return writer->write(const_cast<Bytes*>(&bytes));
685 : : #endif
686 : : }
687 : :
688 : 380 : bool DdsFactory::take_data(dds::DataReader* reader, ReadMessage& msg) {
689 [ + - ]: 380 : auto ret = reader->take(msg.samples, msg.infos, 1);
690 : :
691 : : #ifdef VLINK_SUPPORT_DDS_V3
692 : :
693 : : if (ret == dds::RETCODE_NO_DATA) {
694 : : return false;
695 : : }
696 : :
697 : : if VUNLIKELY (ret != dds::RETCODE_OK) {
698 : : VLOG_E("DdsFactory: Failed to take data.");
699 : : return false;
700 : : }
701 : : #else
702 : :
703 [ + + ]: 380 : if (ret == ReturnCode_t::RETCODE_NO_DATA) {
704 : 190 : return false;
705 : : }
706 : :
707 [ - + ]: 190 : if VUNLIKELY (ret != ReturnCode_t::RETCODE_OK) {
708 [ # # # # ]: 0 : VLOG_E("DdsFactory: Failed to take data.");
709 : 0 : return false;
710 : : }
711 : : #endif
712 : :
713 [ + - ]: 190 : const auto& info = msg.infos[0];
714 : :
715 [ + - ]: 190 : if VLIKELY (info.valid_data) {
716 [ + - + - ]: 190 : msg.id = msg.samples[0].id();
717 : : } else {
718 : 0 : msg.id = 0;
719 : : }
720 : :
721 [ + - ]: 190 : msg.timestamp = info.source_timestamp.to_ns();
722 : :
723 : 190 : return true;
724 : : }
725 : :
726 : 190 : void DdsFactory::return_data_loan(dds::DataReader* reader, ReadMessage& msg) {
727 [ + - ]: 190 : auto ret = reader->return_loan(msg.samples, msg.infos);
728 : :
729 : : #ifdef VLINK_SUPPORT_DDS_V3
730 : :
731 : : if VUNLIKELY (ret != dds::RETCODE_OK) {
732 : : VLOG_E("DdsFactory: Failed to return data loan.");
733 : : }
734 : : #else
735 : :
736 [ - + ]: 190 : if VUNLIKELY (ret != ReturnCode_t::RETCODE_OK) {
737 [ # # # # ]: 0 : VLOG_E("DdsFactory: Failed to return data loan.");
738 : : }
739 : : #endif
740 : 190 : }
741 : :
742 : 0 : bool DdsFactory::take_cdr_data(dds::DataReader* reader, ReadCdrMessage& msg) {
743 [ # # ]: 0 : auto ret = reader->take(msg.samples, msg.infos, 1);
744 : :
745 : : #ifdef VLINK_SUPPORT_DDS_V3
746 : :
747 : : if (ret == dds::RETCODE_NO_DATA) {
748 : : return false;
749 : : }
750 : :
751 : : if VUNLIKELY (ret != dds::RETCODE_OK) {
752 : : VLOG_E("DdsFactory: Failed to take data.");
753 : : return false;
754 : : }
755 : : #else
756 : :
757 [ # # ]: 0 : if (ret == ReturnCode_t::RETCODE_NO_DATA) {
758 : 0 : return false;
759 : : }
760 : :
761 [ # # ]: 0 : if VUNLIKELY (ret != ReturnCode_t::RETCODE_OK) {
762 [ # # # # ]: 0 : VLOG_E("DdsFactory: Failed to take data.");
763 : 0 : return false;
764 : : }
765 : : #endif
766 : :
767 [ # # ]: 0 : const auto& info = msg.infos[0];
768 : 0 : msg.id = info.sample_identity.sequence_number().to64long();
769 : :
770 [ # # ]: 0 : msg.timestamp = info.source_timestamp.to_ns();
771 : :
772 : 0 : return true;
773 : : }
774 : :
775 : 0 : void DdsFactory::return_cdr_loan(dds::DataReader* reader, ReadCdrMessage& msg) {
776 [ # # ]: 0 : auto ret = reader->return_loan(msg.samples, msg.infos);
777 : :
778 : : #ifdef VLINK_SUPPORT_DDS_V3
779 : :
780 : : if VUNLIKELY (ret != dds::RETCODE_OK) {
781 : : VLOG_E("DdsFactory: Failed to return CDR loan.");
782 : : }
783 : : #else
784 : :
785 [ # # ]: 0 : if VUNLIKELY (ret != ReturnCode_t::RETCODE_OK) {
786 [ # # # # ]: 0 : VLOG_E("DdsFactory: Failed to return CDR loan.");
787 : : }
788 : : #endif
789 : 0 : }
790 : :
791 : 27 : uint64_t DdsFactory::get_guid(const rtps::GUID_t& guid, uint32_t seq) {
792 : 27 : const auto& handle = static_cast<const rtps::InstanceHandle_t&>(guid);
793 : 27 : uint64_t result = 14695981039346656037ULL;
794 : :
795 [ + + ]: 459 : for (size_t i = 0; i < 16U; ++i) {
796 : 432 : result ^= static_cast<uint64_t>(handle.value[i]);
797 : 432 : result *= 1099511628211ULL;
798 : : }
799 : :
800 [ + + ]: 135 : for (size_t i = 0; i < sizeof(seq); ++i) {
801 : 108 : result ^= static_cast<uint64_t>((seq >> (i * 8)) & 0xFFU);
802 : 108 : result *= 1099511628211ULL;
803 : : }
804 : :
805 : 27 : return result;
806 : : }
807 : :
808 : 13 : int DdsFactory::get_default_domain_id() {
809 [ + - + - ]: 13 : const std::string& domain_str = Utils::get_env("VLINK_DDS_DOMAIN");
810 : 26 : return Helpers::to_int(domain_str, 0);
811 : 13 : }
812 : :
813 : 129 : void DdsFactory::set_participant_qos(dds::DomainParticipantQos& dds_qos, const Conf::PropertiesMap& properties) {
814 [ + + + - : 129 : static const std::string& ip_str = Utils::get_env("VLINK_DDS_IP");
+ - + - -
- ]
815 [ + + + - : 129 : static const std::string& ip_multicast_str = Utils::get_env("VLINK_DDS_MULTICAST_IP");
+ - + - -
- ]
816 [ + + + - : 129 : static const std::string& peer_str = Utils::get_env("VLINK_DDS_PEER");
+ - + - -
- ]
817 [ + + + - : 129 : static const std::string& buf_str = Utils::get_env("VLINK_DDS_BUF");
+ - + - -
- ]
818 [ + + + - : 129 : static const std::string& mtu_str = Utils::get_env("VLINK_DDS_MTU");
+ - + - -
- ]
819 : :
820 [ + + + - : 129 : static bool enable_udp = Helpers::to_int(Utils::get_env("VLINK_DDS_UDP"), 1) != 0;
+ - + - -
- ]
821 [ + + + - : 129 : static bool enable_tcp = Helpers::to_int(Utils::get_env("VLINK_DDS_TCP"), 0) != 0;
+ - + - -
- ]
822 [ + + + - : 129 : static bool enable_shm = Helpers::to_int(Utils::get_env("VLINK_DDS_SHM"), 0) != 0;
+ - + - -
- ]
823 : :
824 [ + + + - : 129 : static bool enable_less_memory = Helpers::to_int(Utils::get_env("VLINK_DDS_LESS_MEMORY"), 0) != 0;
+ - + - -
- ]
825 : :
826 [ + + + - : 129 : static bool enable_ip_filter = Helpers::to_int(Utils::get_env("VLINK_DDS_IP_FILTER"), 0) != 0;
+ - + - -
- ]
827 : :
828 [ + + + - ]: 129 : static std::vector<std::string> default_ip_list = Utils::get_dds_default_address(enable_ip_filter);
829 : :
830 [ + - ]: 129 : std::string prop_ip_str = ip_str;
831 [ + - ]: 129 : std::string prop_ip_multicast_str = ip_multicast_str;
832 [ + - ]: 129 : std::string prop_peer_str = peer_str;
833 : 129 : size_t prop_buf = 0;
834 : 129 : size_t prop_mtu = 0;
835 : 129 : bool prop_enable_udp = enable_udp;
836 : 129 : bool prop_enable_tcp = enable_tcp;
837 : 129 : [[maybe_unused]] bool prop_enable_shm = enable_shm;
838 : 129 : [[maybe_unused]] bool prop_enable_less_memory = enable_less_memory;
839 : :
840 [ - + ]: 129 : if (!buf_str.empty()) {
841 [ # # ]: 0 : std::from_chars(buf_str.data(), buf_str.data() + buf_str.size(), prop_buf);
842 : : }
843 : :
844 [ - + ]: 129 : if (!mtu_str.empty()) {
845 [ # # ]: 0 : std::from_chars(mtu_str.data(), mtu_str.data() + mtu_str.size(), prop_mtu);
846 : : }
847 : :
848 [ + + ]: 185 : for (const auto& [prop, value] : properties) {
849 [ + + ]: 56 : if (!Helpers::has_startwith(prop, "dds.")) {
850 : 10 : continue;
851 : : }
852 : :
853 [ + + ]: 46 : if (prop == "dds.ip") {
854 [ + - ]: 8 : prop_ip_str = value;
855 [ + + ]: 38 : } else if (prop == "dds.multicast.ip") {
856 [ + - ]: 3 : prop_ip_multicast_str = value;
857 [ + + ]: 35 : } else if (prop == "dds.peer") {
858 [ + - ]: 3 : prop_peer_str = value;
859 [ + + ]: 32 : } else if (prop == "dds.buf") {
860 [ + - ]: 7 : std::from_chars(value.data(), value.data() + value.size(), prop_buf);
861 [ + + ]: 25 : } else if (prop == "dds.mtu") {
862 [ + - ]: 7 : std::from_chars(value.data(), value.data() + value.size(), prop_mtu);
863 [ + + ]: 18 : } else if (prop == "dds.udp") {
864 : 5 : prop_enable_udp = (value == "1");
865 [ + + ]: 13 : } else if (prop == "dds.tcp") {
866 : 6 : prop_enable_tcp = (value == "1");
867 [ + + ]: 7 : } else if (prop == "dds.shm") {
868 : 3 : prop_enable_shm = (value == "1");
869 [ + + ]: 4 : } else if (prop == "dds.less_memory") {
870 : 3 : prop_enable_less_memory = (value == "1");
871 : : } else {
872 [ + - ]: 1 : dds_qos.properties().properties().emplace_back(prop, value);
873 : : }
874 : : }
875 : :
876 : 129 : dds_qos.transport().use_builtin_transports = false;
877 : : #if defined(VLINK_SUPPORT_DDS_V3) || FASTRTPS_VERSION_MINOR >= 10
878 : 129 : dds_qos.wire_protocol().ignore_non_matching_locators = true;
879 : : #endif
880 : 129 : dds_qos.wire_protocol().builtin.avoid_builtin_multicast = true;
881 : :
882 : : // dds_qos.wire_protocol().port.domainIDGain = 250;
883 : : // dds_qos.wire_protocol().port.participantIDGain = 2;
884 : : // dds_qos.wire_protocol().port.portBase = 7400;
885 : : // dds_qos.wire_protocol().port.offsetd0 = 0;
886 : : // dds_qos.wire_protocol().port.offsetd1 = 10;
887 : : // dds_qos.wire_protocol().port.offsetd2 = 1;
888 : : // dds_qos.wire_protocol().port.offsetd3 = 11;
889 : : // #ifdef VLINK_SUPPORT_DDS_V3
890 : : // dds_qos.wire_protocol().builtin.discovery_config.discoveryProtocol = rtps::DiscoveryProtocol::SIMPLE;
891 : : // #else
892 : : // dds_qos.wire_protocol().builtin.discovery_config.discoveryProtocol = rtps::DiscoveryProtocol_t::SIMPLE;
893 : : // #endif
894 : : // dds_qos.wire_protocol().builtin.discovery_config.use_SIMPLE_EndpointDiscoveryProtocol = true;
895 : : // dds_qos.wire_protocol().builtin.discovery_config.m_simpleEDP.use_PublicationReaderANDSubscriptionWriter = true;
896 : : // dds_qos.wire_protocol().builtin.discovery_config.m_simpleEDP.use_PublicationWriterANDSubscriptionReader = true;
897 : : // dds_qos.wire_protocol().builtin.discovery_config.leaseDuration = get_dds_duration(20000);
898 : : // dds_qos.wire_protocol().builtin.discovery_config.leaseDuration_announcementperiod = get_dds_duration(3000);
899 : : // dds_qos.wire_protocol().builtin.discovery_config.initial_announcements.count = 5;
900 : : // dds_qos.wire_protocol().builtin.discovery_config.initial_announcements.period = get_dds_duration(100);
901 : : // dds_qos.wire_protocol().builtin.discovery_config.ignoreParticipantFlags =
902 : : // static_cast<rtps::ParticipantFilteringFlags>(rtps::ParticipantFilteringFlags::FILTER_SAME_PROCESS);
903 : :
904 [ + + ]: 129 : if (prop_enable_less_memory) {
905 : 3 : dds_qos.wire_protocol().builtin.readerHistoryMemoryPolicy = rtps::DYNAMIC_RESERVE_MEMORY_MODE;
906 : 3 : dds_qos.wire_protocol().builtin.writerHistoryMemoryPolicy = rtps::DYNAMIC_RESERVE_MEMORY_MODE;
907 : : // dds_qos.allocation().participants = {2, 10, 1};
908 : : // dds_qos.allocation().readers = {2, 20, 1};
909 : : // dds_qos.allocation().writers = {2, 20, 1};
910 : : // dds_qos.wire_protocol().builtin.readerPayloadSize = 512;
911 : : // dds_qos.wire_protocol().builtin.writerPayloadSize = 512;
912 : : // dds_qos.allocation().locators.max_unicast_locators = 4;
913 : : // dds_qos.allocation().locators.max_multicast_locators = 1;
914 : : // dds_qos.allocation().data_limits.max_user_data = 256;
915 : : // dds_qos.allocation().data_limits.max_properties = 512;
916 : : // dds_qos.allocation().data_limits.max_partitions = 256;
917 : : }
918 : :
919 : : #if !defined(__ANDROID__) && !defined(_WIN32)
920 : :
921 [ - + ]: 129 : if (prop_enable_shm) {
922 [ # # ]: 0 : auto shm_descriptor = std::make_shared<rtps2::SharedMemTransportDescriptor>();
923 [ # # ]: 0 : dds_qos.transport().user_transports.emplace_back(std::move(shm_descriptor));
924 : 0 : }
925 : : #endif
926 : :
927 : : {
928 : 129 : rtps::Locator_t pdp_locator;
929 : 129 : pdp_locator.kind = LOCATOR_KIND_UDPv4;
930 [ + - + - ]: 129 : rtps::IPLocator::setIPv4(pdp_locator, "239.255.0.1");
931 [ + - ]: 129 : dds_qos.wire_protocol().builtin.metatrafficMulticastLocatorList.push_back(std::move(pdp_locator));
932 : : }
933 : :
934 : 129 : std::vector<std::string> ip_str_list;
935 : :
936 [ - + ]: 129 : if (prop_ip_str.empty()) {
937 [ # # ]: 0 : ip_str_list = default_ip_list;
938 : : } else {
939 : 129 : ip_str_list = Helpers::split_any(prop_ip_str);
940 : : }
941 : :
942 [ + - ]: 129 : rtps::LocatorList_t ip_locators = get_locators(ip_str_list);
943 : :
944 [ + - ]: 129 : if (!ip_locators.empty()) {
945 [ + - ]: 129 : if (!prop_enable_shm) {
946 [ + - ]: 129 : dds_qos.wire_protocol().default_unicast_locator_list.push_back(ip_locators);
947 : : }
948 : :
949 [ + - ]: 129 : dds_qos.wire_protocol().builtin.metatrafficUnicastLocatorList.push_back(ip_locators);
950 : : }
951 : :
952 [ + + ]: 129 : if (!prop_ip_multicast_str.empty()) {
953 : 3 : ip_str_list = Helpers::split_any(prop_ip_multicast_str);
954 [ + - ]: 3 : rtps::LocatorList_t multicast_ip_locators = get_locators(ip_str_list);
955 : :
956 [ + - ]: 3 : if (!multicast_ip_locators.empty()) {
957 [ + - ]: 3 : dds_qos.wire_protocol().default_multicast_locator_list.push_back(std::move(multicast_ip_locators));
958 : : }
959 : 3 : }
960 : :
961 [ + + ]: 129 : if (!prop_peer_str.empty()) {
962 : 3 : auto peer_str_list = Helpers::split_any(prop_peer_str);
963 [ + - ]: 3 : rtps::LocatorList_t peer_locators = get_locators(peer_str_list);
964 : :
965 [ + - ]: 3 : if (!peer_locators.empty()) {
966 [ - + ]: 3 : if (dds_qos.wire_protocol().builtin.metatrafficUnicastLocatorList.empty()) {
967 [ # # ]: 0 : dds_qos.wire_protocol().builtin.metatrafficUnicastLocatorList.push_back(rtps::LocatorList_t());
968 : : }
969 : :
970 [ + - ]: 3 : dds_qos.wire_protocol().builtin.initialPeersList.push_back(std::move(peer_locators));
971 : : }
972 : 3 : }
973 : :
974 [ + + ]: 129 : if (prop_enable_udp) {
975 [ + - ]: 127 : auto udp_descriptor = std::make_shared<rtps2::UDPv4TransportDescriptor>();
976 : :
977 [ - + - - : 127 : if (ip_str_list.size() == 1 || !prop_ip_str.empty()) {
+ - ]
978 : : #ifdef VLINK_SUPPORT_DDS_V3
979 : : udp_descriptor->interface_allowlist.reserve(ip_str_list.size());
980 : :
981 : : for (const auto& ip : ip_str_list) {
982 : : // rtps::NetmaskFilterKind::OFF
983 : : udp_descriptor->interface_allowlist.emplace_back(ip);
984 : : }
985 : : #else
986 [ + - ]: 127 : udp_descriptor->interfaceWhiteList = ip_str_list;
987 : : #endif
988 : : }
989 : :
990 [ + + ]: 127 : if (prop_buf > 0) {
991 : 6 : udp_descriptor->sendBufferSize = static_cast<uint32_t>(prop_buf);
992 : 6 : udp_descriptor->receiveBufferSize = static_cast<uint32_t>(prop_buf);
993 : : }
994 : :
995 [ + + ]: 127 : if (prop_mtu > 0) {
996 : 6 : udp_descriptor->maxMessageSize = static_cast<uint32_t>(prop_mtu);
997 : : }
998 : :
999 [ + - ]: 127 : dds_qos.transport().user_transports.emplace_back(std::move(udp_descriptor));
1000 : 127 : }
1001 : :
1002 : 129 : auto ssl_cfg = SslOptions::parse_from(properties);
1003 : :
1004 : 129 : bool ssl_cfg_valid = ssl_cfg.is_valid();
1005 : :
1006 [ + + + - ]: 129 : if (ssl_cfg_valid && !prop_enable_tcp) {
1007 : 2 : prop_enable_tcp = true;
1008 : : }
1009 : :
1010 [ + + ]: 129 : if (prop_enable_tcp) {
1011 [ + - ]: 2 : auto tcp_descriptor = std::make_shared<rtps2::TCPv4TransportDescriptor>();
1012 : :
1013 [ + + ]: 2 : if (prop_buf > 0) {
1014 : 1 : tcp_descriptor->sendBufferSize = static_cast<uint32_t>(prop_buf);
1015 : 1 : tcp_descriptor->receiveBufferSize = static_cast<uint32_t>(prop_buf);
1016 : : }
1017 : :
1018 : 2 : tcp_descriptor->keep_alive_frequency_ms = 1000;
1019 : 2 : tcp_descriptor->keep_alive_timeout_ms = 3000;
1020 : :
1021 [ + - ]: 2 : if (ssl_cfg_valid) {
1022 : 2 : tcp_descriptor->apply_security = true;
1023 : :
1024 [ + - ]: 2 : if (!ssl_cfg.ca_file.empty()) {
1025 [ + - ]: 2 : tcp_descriptor->tls_config.verify_file = ssl_cfg.ca_file;
1026 : : }
1027 : :
1028 [ + + ]: 2 : if (!ssl_cfg.cert_file.empty()) {
1029 [ + - ]: 1 : tcp_descriptor->tls_config.cert_chain_file = ssl_cfg.cert_file;
1030 : : }
1031 : :
1032 [ + + ]: 2 : if (!ssl_cfg.key_file.empty()) {
1033 [ + - ]: 1 : tcp_descriptor->tls_config.private_key_file = ssl_cfg.key_file;
1034 : : }
1035 : :
1036 [ + + ]: 2 : if (!ssl_cfg.key_password.empty()) {
1037 [ + - ]: 1 : tcp_descriptor->tls_config.password = ssl_cfg.key_password;
1038 : : }
1039 : :
1040 : : #if defined(VLINK_SUPPORT_DDS_V3) || FASTRTPS_VERSION_MINOR >= 10
1041 [ + + ]: 2 : if (!ssl_cfg.server_name.empty()) {
1042 [ + - ]: 1 : tcp_descriptor->tls_config.server_name = ssl_cfg.server_name;
1043 : : }
1044 : : #endif
1045 : :
1046 [ + + ]: 2 : if (ssl_cfg.verify_peer) {
1047 : 1 : tcp_descriptor->tls_config.add_verify_mode(
1048 : : rtps2::TCPTransportDescriptor::TLSConfig::TLSVerifyMode::VERIFY_PEER);
1049 : : } else {
1050 : 1 : tcp_descriptor->tls_config.add_verify_mode(
1051 : : rtps2::TCPTransportDescriptor::TLSConfig::TLSVerifyMode::VERIFY_NONE);
1052 : : }
1053 : :
1054 : 2 : tcp_descriptor->tls_config.add_option(rtps2::TCPTransportDescriptor::TLSConfig::TLSOptions::DEFAULT_WORKAROUNDS);
1055 : 2 : tcp_descriptor->tls_config.add_option(rtps2::TCPTransportDescriptor::TLSConfig::TLSOptions::NO_SSLV2);
1056 : 2 : tcp_descriptor->tls_config.add_option(rtps2::TCPTransportDescriptor::TLSConfig::TLSOptions::NO_SSLV3);
1057 : : }
1058 : :
1059 [ + - ]: 2 : tcp_descriptor->add_listener_port(0);
1060 : 2 : rtps::Locator_t tcp_locator;
1061 : 2 : tcp_locator.kind = LOCATOR_KIND_TCPv4;
1062 [ + - + - ]: 2 : rtps::IPLocator::setIPv4(tcp_locator, "0.0.0.0");
1063 [ + - ]: 2 : rtps::IPLocator::setPhysicalPort(tcp_locator, 0);
1064 [ + - ]: 2 : rtps::IPLocator::setLogicalPort(tcp_locator, 0);
1065 [ + - ]: 2 : dds_qos.wire_protocol().builtin.metatrafficUnicastLocatorList.push_back(tcp_locator);
1066 [ + - ]: 2 : dds_qos.wire_protocol().default_unicast_locator_list.push_back(tcp_locator);
1067 : 2 : rtps::Locator_t pdp_locator;
1068 : 2 : pdp_locator.kind = LOCATOR_KIND_UDPv4;
1069 [ + - + - ]: 2 : rtps::IPLocator::setIPv4(pdp_locator, "239.255.0.1");
1070 [ + - ]: 2 : dds_qos.wire_protocol().builtin.metatrafficMulticastLocatorList.push_back(std::move(pdp_locator));
1071 [ + - ]: 2 : dds_qos.transport().user_transports.emplace_back(std::move(tcp_descriptor));
1072 : 2 : }
1073 : 129 : }
1074 : :
1075 : 135 : rtps::LocatorList_t DdsFactory::get_locators(const std::vector<std::string>& list) {
1076 : 135 : rtps::LocatorList_t locator_list;
1077 : :
1078 [ - + ]: 135 : if (list.empty()) {
1079 : 0 : return locator_list;
1080 : : }
1081 : :
1082 [ + + ]: 270 : for (const auto& ip : list) {
1083 [ - + ]: 135 : if (ip.find(':') != std::string::npos) {
1084 : 0 : rtps2::Locator locator;
1085 : 0 : locator.kind = LOCATOR_KIND_UDPv6;
1086 : 0 : locator.port = 0;
1087 : :
1088 [ # # # # ]: 0 : if VLIKELY (rtps::IPLocator::setIPv6(locator, ip)) {
1089 [ # # ]: 0 : locator_list.push_back(std::move(locator));
1090 : : }
1091 : : } else {
1092 : 135 : rtps2::Locator locator;
1093 : 135 : locator.kind = LOCATOR_KIND_UDPv4;
1094 : 135 : locator.port = 0;
1095 : :
1096 [ + - + - ]: 135 : if VLIKELY (rtps::IPLocator::setIPv4(locator, ip)) {
1097 [ + - ]: 135 : locator_list.push_back(std::move(locator));
1098 : : }
1099 : : }
1100 : : }
1101 : :
1102 : 135 : return locator_list;
1103 : 0 : }
1104 : :
1105 : 1454 : std::string DdsFactory::get_qos_ext(const Conf::PropertiesMap& ext, const std::string& key) {
1106 [ + - ]: 1454 : auto iter = ext.find(key);
1107 : :
1108 [ + + ]: 1454 : if (iter == ext.end()) {
1109 [ + - ]: 1447 : return "";
1110 : : }
1111 : :
1112 [ + - ]: 7 : return iter->second;
1113 : : }
1114 : :
1115 : : } // namespace vlink
|
