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