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_client_impl.hpp"
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 : : // WriterListener
35 : 29 : DdscClientImpl::WriterListener::WriterListener(NodeImpl* impl) : DdscWriterListener(impl) {}
36 : :
37 : 22 : void DdscClientImpl::WriterListener::on_publication_matched(NodeImpl* impl,
38 : : const dds_publication_matched_status_t& status) {
39 : 22 : auto* instance = static_cast<DdscClientImpl*>(impl);
40 : 22 : auto* message_loop = instance->get_message_loop();
41 : :
42 : 22 : instance->write_session_count_.store(status.current_count, std::memory_order_release);
43 : :
44 [ + + ]: 22 : if (message_loop) {
45 [ + - + - ]: 1 : message_loop->post_task([instance]() {
46 [ - + ]: 1 : if VUNLIKELY (!instance->get_message_loop()) {
47 : 0 : return;
48 : : }
49 : :
50 : 1 : instance->update_connected();
51 : : });
52 : : } else {
53 : 21 : instance->update_connected();
54 : : }
55 : :
56 : 22 : DdscWriterListener::on_publication_matched(impl, status);
57 : 22 : }
58 : :
59 : : // ReaderListener
60 : 22 : DdscClientImpl::ReaderListener::ReaderListener(NodeImpl* impl) : DdscReaderListener(impl) {}
61 : :
62 : 20 : void DdscClientImpl::ReaderListener::on_subscription_matched(NodeImpl* impl,
63 : : const dds_subscription_matched_status_t& status) {
64 : 20 : auto* instance = static_cast<DdscClientImpl*>(impl);
65 : 20 : auto* message_loop = instance->get_message_loop();
66 : :
67 : 20 : instance->read_session_count_.store(status.current_count, std::memory_order_release);
68 : :
69 [ + + ]: 20 : if (message_loop) {
70 [ + - + - ]: 1 : message_loop->post_task([instance]() {
71 [ - + ]: 1 : if VUNLIKELY (!instance->get_message_loop()) {
72 : 0 : return;
73 : : }
74 : :
75 : 1 : instance->update_connected();
76 : : });
77 : : } else {
78 : 19 : instance->update_connected();
79 : : }
80 : :
81 : 20 : DdscReaderListener::on_subscription_matched(impl, status);
82 : 20 : }
83 : :
84 : 26 : void DdscClientImpl::ReaderListener::on_data_available(NodeImpl* impl, dds_entity_t reader) {
85 : 26 : auto* instance = static_cast<DdscClientImpl*>(impl);
86 : 26 : auto* message_loop = instance->get_message_loop();
87 : :
88 [ + + ]: 26 : if (message_loop) {
89 [ + - + - ]: 2 : message_loop->post_task([instance, reader]() {
90 [ - + ]: 2 : if VUNLIKELY (!instance->get_message_loop()) {
91 : 0 : return;
92 : : }
93 : :
94 : 2 : instance->process_message(reader);
95 : : });
96 : : } else {
97 : 24 : instance->process_message(reader);
98 : : }
99 : 26 : }
100 : :
101 : : // DdscClientImpl
102 [ + - + - ]: 33 : DdscClientImpl::DdscClientImpl(const DdscConf& conf) : conf_(conf) {}
103 : :
104 : 26 : void DdscClientImpl::process_message(dds_entity_t reader) {
105 : 26 : DdscFactory::ReadMessage msg;
106 : :
107 [ + - + + ]: 52 : while (DdscFactory::take_data(reader, msg)) {
108 [ - + ]: 26 : if VUNLIKELY (quit_flag_.load(std::memory_order_acquire)) {
109 [ # # ]: 0 : DdscFactory::release_data(reader, msg);
110 : 0 : break;
111 : : }
112 : :
113 [ - + ]: 26 : if VUNLIKELY (!msg.info.valid_data) {
114 [ # # ]: 0 : DdscFactory::release_data(reader, msg);
115 : 6 : continue;
116 : : }
117 : :
118 : 26 : NodeImpl::MsgCallback cb;
119 : : {
120 [ + - ]: 26 : std::lock_guard param_lock(param_mtx_);
121 [ + - ]: 26 : auto iter = callbacks_.find(msg.id);
122 : :
123 [ + + ]: 26 : if VUNLIKELY (iter == callbacks_.end()) {
124 [ + - ]: 6 : DdscFactory::release_data(reader, msg);
125 : 6 : continue;
126 : : }
127 : :
128 : 20 : cb = std::move(iter->second);
129 [ + - ]: 20 : callbacks_.erase(iter);
130 [ + + ]: 26 : }
131 : :
132 [ + - ]: 20 : if VLIKELY (cb) {
133 [ + - ]: 20 : cb(msg.bytes);
134 : : }
135 : :
136 [ + - ]: 20 : DdscFactory::release_data(reader, msg);
137 [ + + ]: 26 : }
138 : 26 : }
139 : :
140 : 29 : void DdscClientImpl::init() {
141 [ + - ]: 29 : participant_ = DdscFactory::create_participant(kServer | kClient, conf_, get_all_properties());
142 : :
143 [ - + ]: 29 : if VUNLIKELY (!participant_) {
144 [ # # # # ]: 0 : VLOG_E("DdscClientImpl::init(): participant creation failed; client left uninitialised.");
145 : :
146 : 0 : return;
147 : : }
148 : :
149 [ + + ]: 29 : if (is_resp_type) {
150 : 22 : std::tie(topic_req_, topic_resp_) = DdscFactory::create_method_topic(kServer | kClient, conf_, participant_.get());
151 : :
152 : 22 : publisher_ = DdscFactory::create_publisher(kClient, conf_, participant_.get());
153 : :
154 [ + - ]: 22 : writer_listener_.emplace(this);
155 : :
156 : : writer_ =
157 : 22 : DdscFactory::create_datawriter(kClient, conf_, publisher_.get(), topic_req_.get(), writer_listener_->get_ptr());
158 : :
159 : 22 : subscriber_ = DdscFactory::create_subscriber(kClient, conf_, participant_.get());
160 : :
161 [ + - ]: 22 : reader_listener_.emplace(this);
162 : :
163 : 44 : reader_ = DdscFactory::create_datareader(kClient, conf_, subscriber_.get(), topic_resp_.get(),
164 : 44 : reader_listener_->get_ptr());
165 : : } else {
166 [ + - + - ]: 7 : topic_req_ = DdscFactory::create_topic(kServer | kClient, conf_, participant_.get());
167 : :
168 : 7 : publisher_ = DdscFactory::create_publisher(kClient, conf_, participant_.get());
169 : :
170 [ + - ]: 7 : writer_listener_.emplace(this);
171 : :
172 : : writer_ =
173 : 7 : DdscFactory::create_datawriter(kClient, conf_, publisher_.get(), topic_req_.get(), writer_listener_->get_ptr());
174 : : }
175 : :
176 : 29 : quit_flag_.store(false, std::memory_order_release);
177 : : }
178 : :
179 : 29 : void DdscClientImpl::deinit() {
180 : 29 : quit_flag_.store(true, std::memory_order_release);
181 : :
182 : 29 : detach();
183 : :
184 : 29 : reader_.reset();
185 : 29 : writer_.reset();
186 : 29 : reader_listener_.reset();
187 : 29 : writer_listener_.reset();
188 : 29 : subscriber_.reset();
189 : 29 : publisher_.reset();
190 : 29 : topic_resp_.reset();
191 : 29 : topic_req_.reset();
192 : 29 : participant_.reset();
193 : 29 : }
194 : :
195 : 29 : void DdscClientImpl::interrupt() {
196 : 29 : ClientImpl::interrupt();
197 : :
198 : 29 : ack_manager_.clear();
199 : 29 : }
200 : :
201 : 0 : const Conf* DdscClientImpl::get_conf() const { return &conf_; }
202 : :
203 : 4 : const AbstractNode* DdscClientImpl::get_abstract_node() const { return this; }
204 : :
205 : 7 : Status::BasePtr DdscClientImpl::get_status(Status::Type type) const {
206 [ + + ]: 7 : if (Status::is_for_writer(type)) {
207 [ + + ]: 3 : if VUNLIKELY (!writer_) {
208 : 1 : return std::make_shared<Status::Unknown>();
209 : : }
210 : :
211 : 2 : return WriterListener::get_status(writer_->entity, type);
212 : : }
213 : :
214 [ + + ]: 4 : if VUNLIKELY (!reader_) {
215 : 2 : return std::make_shared<Status::Unknown>();
216 : : }
217 : :
218 [ + - ]: 2 : if (reader_listener_) {
219 : 2 : return ReaderListener::get_status(reader_->entity, type);
220 : : }
221 : :
222 : 0 : return std::make_shared<Status::Unknown>();
223 : : }
224 : :
225 : 2 : std::any DdscClientImpl::get_native_handle() const { return publisher_; }
226 : :
227 : 93 : bool DdscClientImpl::is_connected() const {
228 [ + + ]: 93 : if (is_resp_type) {
229 [ + + ]: 231 : return write_session_count_.load(std::memory_order_acquire) > 0 &&
230 [ + + ]: 222 : read_session_count_.load(std::memory_order_acquire) > 0;
231 : : } else {
232 : 26 : return write_session_count_.load(std::memory_order_acquire) > 0;
233 : : }
234 : : }
235 : :
236 : 31 : bool DdscClientImpl::call(const Bytes& req_data, MsgCallback&& callback, std::chrono::milliseconds timeout) {
237 [ + + ]: 31 : if (!callback) {
238 [ + - ]: 6 : return DdscFactory::write_data(writer_->entity, req_data, 0);
239 : : }
240 : :
241 : : dds_guid_t guid;
242 [ + - ]: 25 : dds_get_guid(writer_->entity, &guid);
243 [ + - ]: 50 : uint64_t id = DdscFactory::get_guid(&guid, seq_.fetch_add(1, std::memory_order_relaxed) + 1);
244 : :
245 : 10 : auto cleanup_callback = [this, &id]() {
246 [ + - ]: 5 : std::lock_guard param_lock(param_mtx_);
247 [ + - ]: 5 : callbacks_.erase(id);
248 : 5 : };
249 : :
250 [ + + ]: 25 : if (timeout.count() != 0) {
251 : 19 : ack_manager_.reset_interrupted();
252 : :
253 : 19 : auto ack_request = ack_manager_.create_request();
254 : :
255 : : {
256 [ + - ]: 19 : std::lock_guard param_lock(param_mtx_);
257 [ + - ]: 38 : callbacks_[id] = [this, ack_request, callback = std::move(callback)](const Bytes& resp_data) {
258 [ + - ]: 28 : ack_manager_.notify(ack_request, [&callback, &resp_data]() { callback(resp_data); });
259 [ + - ]: 71 : };
260 : 19 : }
261 : :
262 : 19 : ElapsedTimer timer;
263 : 19 : timer.start();
264 : :
265 [ + - + + ]: 19 : if VUNLIKELY (!wait_for_connected(timeout)) {
266 [ + - ]: 1 : cleanup_callback();
267 : 1 : return false;
268 : : }
269 : :
270 : 18 : auto elapsed = timer.get();
271 : :
272 [ + - - + : 18 : if VUNLIKELY (timeout.count() > 0 && elapsed >= timeout.count()) {
- + ]
273 [ # # ]: 0 : cleanup_callback();
274 : 0 : return false;
275 : : }
276 : :
277 [ + - ]: 18 : bool result = ack_manager_.process(ack_request, timeout.count() - elapsed, [this, &req_data, &id]() {
278 : 18 : return DdscFactory::write_data(writer_->entity, req_data, id);
279 : : });
280 : :
281 [ + + ]: 18 : if VUNLIKELY (!result) {
282 [ + - ]: 4 : cleanup_callback();
283 : : }
284 : :
285 : 18 : return result;
286 : 19 : }
287 : :
288 : : {
289 [ + - ]: 6 : std::lock_guard param_lock(param_mtx_);
290 [ + - + - ]: 12 : callbacks_[id] = [callback = std::move(callback)](const Bytes& resp_data) { callback(resp_data); };
291 : 6 : }
292 : :
293 [ + - ]: 6 : bool result = DdscFactory::write_data(writer_->entity, req_data, id);
294 : :
295 [ - + ]: 6 : if VUNLIKELY (!result) {
296 [ # # ]: 0 : cleanup_callback();
297 : : }
298 : :
299 : 6 : return result;
300 : : }
301 : :
302 : : } // namespace vlink
|
