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