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_subscriber_impl.h"
25 : :
26 : : #include <memory>
27 : : #include <utility>
28 : :
29 : : #include "./base/elapsed_timer.h"
30 : : #include "./base/message_loop.h"
31 : :
32 : : namespace vlink {
33 : :
34 : : // ReaderListener
35 : 60 : DdsSubscriberImpl::ReaderListener::ReaderListener(NodeImpl* impl) : DdsReaderListener(impl) {}
36 : :
37 : 107 : void DdsSubscriberImpl::ReaderListener::on_data_available(dds::DataReader* reader) {
38 : 107 : auto* instance = static_cast<DdsSubscriberImpl*>(get_impl());
39 : 107 : auto* message_loop = instance->get_message_loop();
40 : :
41 [ + + ]: 107 : if VUNLIKELY (instance->has_suspend.load(std::memory_order_relaxed)) {
42 [ - + ]: 1 : if (instance->is_cdr_type) {
43 : 0 : DdsFactory::ReadCdrMessage msg;
44 : :
45 [ # # ]: 0 : msg.sample = instance->type_support_.create_data();
46 : :
47 [ # # # # ]: 0 : while (DdsFactory::take_cdr_data(reader, msg)) {
48 [ # # ]: 0 : if VUNLIKELY (instance->quit_flag_.load(std::memory_order_acquire)) {
49 : 0 : break;
50 : : }
51 : : }
52 : :
53 [ # # ]: 0 : instance->type_support_.delete_data(msg.sample);
54 : 0 : } else {
55 [ + - ]: 1 : DdsFactory::ReadMessage msg;
56 : :
57 [ + - + + ]: 2 : while (DdsFactory::take_data(reader, msg)) {
58 [ - + ]: 1 : if VUNLIKELY (instance->quit_flag_.load(std::memory_order_acquire)) {
59 : 0 : break;
60 : : }
61 : : }
62 : 1 : }
63 : :
64 : 1 : return;
65 : : }
66 : :
67 [ - + ]: 106 : if VUNLIKELY (!instance->callback_) {
68 : 0 : return;
69 : : }
70 : :
71 [ + + ]: 106 : if (message_loop) {
72 [ + - + - ]: 1 : message_loop->post_task([instance, reader]() {
73 [ - + ]: 1 : if VUNLIKELY (!instance->get_message_loop()) {
74 : 0 : return;
75 : : }
76 : :
77 : 1 : instance->process_message(reader);
78 : : });
79 : : } else {
80 : 105 : instance->process_message(reader);
81 : : }
82 : : }
83 : :
84 : : // DdsSubscriberImpl
85 [ + - + - ]: 73 : DdsSubscriberImpl::DdsSubscriberImpl(const DdsConf& conf) : conf_(conf) {}
86 : :
87 : 106 : void DdsSubscriberImpl::process_message(dds::DataReader* reader) {
88 [ - + ]: 106 : if (is_cdr_type) {
89 : 0 : DdsFactory::ReadCdrMessage msg;
90 : :
91 [ # # ]: 0 : msg.sample = type_support_.create_data();
92 : :
93 [ # # # # ]: 0 : while (DdsFactory::take_cdr_data(reader, msg)) {
94 [ # # ]: 0 : if VUNLIKELY (quit_flag_.load(std::memory_order_acquire)) {
95 : 0 : break;
96 : : }
97 : :
98 [ # # ]: 0 : if VUNLIKELY (!msg.info.valid_data) {
99 : 0 : continue;
100 : : }
101 : :
102 [ # # ]: 0 : if VUNLIKELY (is_latency_and_lost_enabled_.load(std::memory_order_acquire)) {
103 : 0 : last_latency_.store(ElapsedTimer::get_sys_timestamp(ElapsedTimer::kNano, false) - msg.timestamp,
104 : : std::memory_order_relaxed);
105 : :
106 : 0 : uint64_t part1 = 0;
107 : 0 : uint64_t part2 = 0;
108 : :
109 : 0 : std::memcpy(&part1, msg.info.publication_handle.value, sizeof(uint64_t));
110 : 0 : std::memcpy(&part2, msg.info.publication_handle.value + 8, sizeof(uint64_t));
111 : :
112 : 0 : calc_sample_.update(msg.id, part1 ^ part2);
113 : : }
114 : :
115 [ # # ]: 0 : callback_(msg.bytes);
116 : : }
117 : :
118 [ # # ]: 0 : type_support_.delete_data(msg.sample);
119 : 0 : } else {
120 [ + - ]: 106 : DdsFactory::ReadMessage msg;
121 : :
122 [ + - + + ]: 212 : while (DdsFactory::take_data(reader, msg)) {
123 [ - + ]: 106 : if VUNLIKELY (quit_flag_.load(std::memory_order_acquire)) {
124 : 0 : break;
125 : : }
126 : :
127 [ - + ]: 106 : if VUNLIKELY (!msg.info.valid_data) {
128 : 0 : continue;
129 : : }
130 : :
131 [ + + ]: 106 : if VUNLIKELY (is_latency_and_lost_enabled_.load(std::memory_order_acquire)) {
132 : 10 : last_latency_.store(ElapsedTimer::get_sys_timestamp(ElapsedTimer::kNano, false) - msg.timestamp,
133 : : std::memory_order_relaxed);
134 : :
135 : 10 : uint64_t part1 = 0;
136 : 10 : uint64_t part2 = 0;
137 : :
138 : 10 : std::memcpy(&part1, msg.info.publication_handle.value, sizeof(uint64_t));
139 : 10 : std::memcpy(&part2, msg.info.publication_handle.value + 8, sizeof(uint64_t));
140 : :
141 : 10 : calc_sample_.update(msg.id, part1 ^ part2);
142 : : }
143 : :
144 [ + - ]: 106 : callback_(msg.bytes);
145 : : }
146 : 106 : }
147 : 106 : }
148 : :
149 : 69 : void DdsSubscriberImpl::init() {
150 [ - + - - : 69 : if VUNLIKELY (is_cdr_type && is_security_type) {
- + ]
151 [ # # # # ]: 0 : VLOG_F("Cdr type does not support security.");
152 : : }
153 : :
154 [ + - ]: 69 : participant_ = DdsFactory::create_participant(kPublisher | kSubscriber, conf_, get_all_properties());
155 : :
156 [ + - + - ]: 69 : topic_ = DdsFactory::create_topic(kPublisher | kSubscriber, conf_, participant_.get(), is_cdr_type);
157 : :
158 : 69 : subscriber_ = DdsFactory::create_subscriber(kSubscriber, conf_, participant_.get());
159 : :
160 [ + - - + : 69 : if VUNLIKELY (!participant_ || !topic_) {
- + ]
161 [ # # # # ]: 0 : VLOG_E("DdsSubscriberImpl::init(): participant/topic creation failed; subscriber left uninitialised.");
162 : :
163 : 0 : return;
164 : : }
165 : :
166 : 69 : type_support_ = participant_->find_type(topic_->get_type_name());
167 : :
168 [ - + ]: 69 : if VUNLIKELY (!type_support_) {
169 [ # # # # ]: 0 : CLOG_F("Failed to find typesupport (%s).", topic_->get_type_name().c_str());
170 : : }
171 : :
172 : 69 : quit_flag_.store(false, std::memory_order_release);
173 : : }
174 : :
175 : 69 : void DdsSubscriberImpl::deinit() {
176 : 69 : quit_flag_.store(true, std::memory_order_release);
177 : :
178 : 69 : detach();
179 : :
180 : 69 : reader_.reset();
181 : 69 : listener_.reset();
182 : 69 : subscriber_.reset();
183 : 69 : topic_.reset();
184 : 69 : participant_.reset();
185 : 69 : type_support_.reset();
186 : 69 : }
187 : :
188 : 2 : bool DdsSubscriberImpl::suspend() {
189 : 2 : has_suspend.store(true, std::memory_order_relaxed);
190 : :
191 : 2 : return true;
192 : : }
193 : :
194 : 2 : bool DdsSubscriberImpl::resume() {
195 : 2 : has_suspend.store(false, std::memory_order_relaxed);
196 : :
197 : 2 : return true;
198 : : }
199 : :
200 : 3 : bool DdsSubscriberImpl::is_suspend() const { return has_suspend.load(std::memory_order_relaxed); }
201 : :
202 : 0 : const Conf* DdsSubscriberImpl::get_conf() const { return &conf_; }
203 : :
204 : 16 : const AbstractNode* DdsSubscriberImpl::get_abstract_node() const { return this; }
205 : :
206 : 9 : Status::BasePtr DdsSubscriberImpl::get_status(Status::Type type) const {
207 [ + + ]: 9 : if VUNLIKELY (!reader_) {
208 : 2 : return std::make_shared<Status::Unknown>();
209 : : }
210 : :
211 : 7 : return ReaderListener::get_status(reader_.get(), type);
212 : : }
213 : :
214 : 11 : std::any DdsSubscriberImpl::get_native_handle() const { return subscriber_; }
215 : :
216 : 60 : bool DdsSubscriberImpl::listen(MsgCallback&& callback) {
217 [ - + ]: 60 : if VUNLIKELY (callback_) {
218 : 0 : return false;
219 : : }
220 : :
221 : 60 : callback_ = std::move(callback);
222 : :
223 [ + - ]: 60 : listener_.emplace(this);
224 : :
225 : 60 : reader_ = DdsFactory::create_datareader(kSubscriber, conf_, subscriber_.get(), topic_.get(), &listener_.value());
226 : :
227 : 60 : return true;
228 : : }
229 : :
230 : 4 : void DdsSubscriberImpl::set_latency_and_lost_enabled(bool enable) {
231 : 4 : is_latency_and_lost_enabled_.store(enable, std::memory_order_release);
232 : 4 : }
233 : :
234 : 5 : bool DdsSubscriberImpl::is_latency_and_lost_enabled() const {
235 : 5 : return is_latency_and_lost_enabled_.load(std::memory_order_acquire);
236 : : }
237 : :
238 : 3 : int64_t DdsSubscriberImpl::get_latency() const {
239 [ + + ]: 3 : if (!is_latency_and_lost_enabled_.load(std::memory_order_acquire)) {
240 : 1 : return 0;
241 : : }
242 : :
243 : 4 : return last_latency_.load(std::memory_order_relaxed);
244 : : }
245 : :
246 : 3 : SampleLostInfo DdsSubscriberImpl::get_lost() const {
247 [ + + ]: 3 : if (!is_latency_and_lost_enabled_.load(std::memory_order_acquire)) {
248 : 1 : return SampleLostInfo();
249 : : }
250 : :
251 : 2 : return SampleLostInfo{calc_sample_.get_total(), calc_sample_.get_lost()};
252 : : }
253 : :
254 : : } // namespace vlink
|
