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_server_impl.h"
25 : :
26 : : #include <memory>
27 : : #include <utility>
28 : :
29 : : #include "./base/message_loop.h"
30 : :
31 : : namespace vlink {
32 : :
33 : : // WriterListener
34 : 23 : DdsServerImpl::WriterListener::WriterListener(NodeImpl* impl) : DdsWriterListener(impl) {}
35 : :
36 : : // ReaderListener
37 : 28 : DdsServerImpl::ReaderListener::ReaderListener(NodeImpl* impl) : DdsReaderListener(impl) {}
38 : :
39 : 47 : void DdsServerImpl::ReaderListener::on_subscription_matched(dds::DataReader* reader,
40 : : const dds::SubscriptionMatchedStatus& status) {
41 : 47 : auto* instance = static_cast<DdsServerImpl*>(get_impl());
42 : :
43 : 47 : instance->read_session_count_.store(status.current_count, std::memory_order_relaxed);
44 : :
45 : 47 : DdsReaderListener::on_subscription_matched(reader, status);
46 : 47 : }
47 : :
48 : 31 : void DdsServerImpl::ReaderListener::on_data_available(dds::DataReader* reader) {
49 : 31 : auto* instance = static_cast<DdsServerImpl*>(get_impl());
50 : 31 : auto* message_loop = instance->get_message_loop();
51 : :
52 [ + + ]: 31 : if VUNLIKELY (instance->has_suspend.load(std::memory_order_acquire)) {
53 [ - + ]: 1 : if (instance->is_cdr_type) {
54 : 0 : DdsFactory::ReadCdrMessage msg;
55 : :
56 [ # # # # ]: 0 : while (DdsFactory::take_cdr_data(reader, msg)) {
57 : 0 : const bool should_quit = instance->quit_flag_.load(std::memory_order_relaxed);
58 [ # # ]: 0 : DdsFactory::return_cdr_loan(reader, msg);
59 : :
60 [ # # ]: 0 : if VUNLIKELY (should_quit) {
61 : 0 : break;
62 : : }
63 : : }
64 : 0 : } else {
65 : 1 : DdsFactory::ReadMessage msg;
66 : :
67 [ + - + + ]: 2 : while (DdsFactory::take_data(reader, msg)) {
68 [ + - ]: 1 : DdsFactory::return_data_loan(reader, msg);
69 : :
70 [ - + ]: 1 : if VUNLIKELY (instance->quit_flag_.load(std::memory_order_relaxed)) {
71 : 0 : break;
72 : : }
73 : : }
74 : 1 : }
75 : :
76 : 1 : return;
77 : : }
78 : :
79 [ - + ]: 30 : if VUNLIKELY (!instance->callback_) {
80 : 0 : return;
81 : : }
82 : :
83 [ + + ]: 30 : if (message_loop) {
84 [ + - + - ]: 2 : message_loop->post_task([instance, reader]() {
85 [ - + ]: 2 : if VUNLIKELY (!instance->get_message_loop()) {
86 : 0 : return;
87 : : }
88 : :
89 : 2 : instance->process_message(reader);
90 : : });
91 : : } else {
92 : 28 : instance->process_message(reader);
93 : : }
94 : : }
95 : :
96 : : // DdsServerImpl
97 [ + - + - ]: 37 : DdsServerImpl::DdsServerImpl(const DdsConf& conf) : conf_(conf) {}
98 : :
99 : 30 : void DdsServerImpl::process_message(dds::DataReader* reader) {
100 [ - + ]: 30 : if (is_cdr_type) {
101 : 0 : DdsFactory::ReadCdrMessage msg;
102 : :
103 [ # # # # ]: 0 : while (DdsFactory::take_cdr_data(reader, msg)) {
104 [ # # ]: 0 : if VUNLIKELY (quit_flag_.load(std::memory_order_relaxed)) {
105 [ # # ]: 0 : DdsFactory::return_cdr_loan(reader, msg);
106 : 0 : break;
107 : : }
108 : :
109 [ # # ]: 0 : const auto& info = msg.infos[0];
110 : :
111 [ # # ]: 0 : if VUNLIKELY (!info.valid_data) {
112 [ # # ]: 0 : DdsFactory::return_cdr_loan(reader, msg);
113 : 0 : continue;
114 : : }
115 : :
116 [ # # ]: 0 : if (writer_) {
117 : : {
118 [ # # ]: 0 : std::lock_guard lock(param_mtx_);
119 [ # # ]: 0 : rtps::WriteParams param;
120 : 0 : param.related_sample_identity() = info.sample_identity;
121 : 0 : msg.id = cdr_seq_.fetch_add(1, std::memory_order_relaxed) + 1;
122 [ # # ]: 0 : cdr_id_map_.emplace(msg.id, std::move(param));
123 : 0 : }
124 : :
125 : 0 : Bytes resp_data;
126 : :
127 [ # # # # ]: 0 : callback_(msg.id, msg.samples[0], &resp_data);
128 : 0 : } else {
129 [ # # # # ]: 0 : callback_(msg.id, msg.samples[0], nullptr);
130 : : }
131 : :
132 [ # # ]: 0 : DdsFactory::return_cdr_loan(reader, msg);
133 : : }
134 : 0 : } else {
135 : 30 : DdsFactory::ReadMessage msg;
136 : :
137 [ + - + + ]: 60 : while (DdsFactory::take_data(reader, msg)) {
138 [ - + ]: 30 : if VUNLIKELY (quit_flag_.load(std::memory_order_relaxed)) {
139 [ # # ]: 0 : DdsFactory::return_data_loan(reader, msg);
140 : 0 : break;
141 : : }
142 : :
143 [ + - ]: 30 : const auto& info = msg.infos[0];
144 : :
145 [ - + ]: 30 : if VUNLIKELY (!info.valid_data) {
146 [ # # ]: 0 : DdsFactory::return_data_loan(reader, msg);
147 : 0 : continue;
148 : : }
149 : :
150 [ + + ]: 30 : if (writer_) {
151 : 25 : Bytes resp_data;
152 [ + - + - : 25 : callback_(msg.id, msg.samples[0].data(), &resp_data);
+ - ]
153 : 25 : } else {
154 [ + - + - : 5 : callback_(msg.id, msg.samples[0].data(), nullptr);
+ - ]
155 : : }
156 : :
157 [ + - ]: 30 : DdsFactory::return_data_loan(reader, msg);
158 : : }
159 : 30 : }
160 : 30 : }
161 : :
162 : 30 : void DdsServerImpl::init() {
163 [ + + - + : 30 : if VUNLIKELY (is_resp_type && is_cdr_type != is_resp_cdr_type) {
- + ]
164 [ # # # # ]: 0 : VLOG_F("DdsServer: Request and response must both use raw or CDR serialization.");
165 : : }
166 : :
167 [ - + - - : 30 : if VUNLIKELY (is_cdr_type && is_security_type) {
- + ]
168 [ # # # # ]: 0 : VLOG_F("Cdr type does not support security.");
169 : : }
170 : :
171 [ + - ]: 30 : participant_ = DdsFactory::create_participant(kServer | kClient, conf_, get_all_properties());
172 : :
173 [ + + ]: 30 : if (is_resp_type) {
174 : 23 : std::tie(topic_req_, topic_resp_) =
175 : 46 : DdsFactory::create_method_topic(kServer | kClient, conf_, participant_.get(), is_cdr_type, ser_type);
176 : : } else {
177 [ + - ]: 7 : topic_req_ = DdsFactory::create_topic(kServer | kClient, conf_, participant_.get(), is_cdr_type, {}, ser_type);
178 : : }
179 : :
180 [ + - - + : 30 : if VUNLIKELY (!participant_ || !topic_req_ || (is_resp_type && !topic_resp_)) {
+ - + + -
+ - + -
+ ]
181 [ # # # # ]: 0 : VLOG_E("DdsServerImpl::init(): participant/topic creation failed; server left uninitialised.");
182 : 0 : return;
183 : : }
184 : :
185 [ - + ]: 30 : if (is_cdr_type) {
186 : 0 : ser_type = topic_req_->get_type_name();
187 [ # # ]: 0 : if (is_resp_type) {
188 [ # # ]: 0 : ser_type += "|" + topic_resp_->get_type_name();
189 : : }
190 : : }
191 : :
192 [ + + ]: 30 : if (is_resp_type) {
193 : 23 : publisher_ = DdsFactory::create_publisher(kServer, conf_, participant_.get());
194 : :
195 [ + - ]: 23 : writer_listener_.emplace(this);
196 : :
197 : 46 : writer_ = DdsFactory::create_datawriter(kServer, conf_, publisher_.get(), topic_resp_.get(),
198 : 46 : &writer_listener_.value(), is_cdr_type);
199 : : }
200 : :
201 : 30 : subscriber_ = DdsFactory::create_subscriber(kServer, conf_, participant_.get());
202 : :
203 : 30 : quit_flag_.store(false, std::memory_order_relaxed);
204 : : }
205 : :
206 : 30 : void DdsServerImpl::deinit() {
207 : 30 : quit_flag_.store(true, std::memory_order_relaxed);
208 : :
209 [ + - ]: 30 : detach();
210 : :
211 : 30 : reader_.reset();
212 : 30 : writer_.reset();
213 : 30 : reader_listener_.reset();
214 : 30 : writer_listener_.reset();
215 : 30 : subscriber_.reset();
216 : 30 : publisher_.reset();
217 : 30 : topic_resp_.reset();
218 : 30 : topic_req_.reset();
219 : 30 : participant_.reset();
220 [ + - ]: 30 : std::lock_guard lock(param_mtx_);
221 : 30 : cdr_id_map_.clear();
222 : 30 : callback_ = {};
223 : 30 : is_listened = false;
224 : 30 : read_session_count_.store(0, std::memory_order_relaxed);
225 : 30 : }
226 : :
227 : 3 : bool DdsServerImpl::suspend() {
228 : 3 : has_suspend.store(true, std::memory_order_release);
229 : :
230 : 3 : return true;
231 : : }
232 : :
233 : 3 : bool DdsServerImpl::resume() {
234 : 3 : has_suspend.store(false, std::memory_order_release);
235 : :
236 : 3 : return true;
237 : : }
238 : :
239 : 6 : bool DdsServerImpl::is_suspend() const { return has_suspend.load(std::memory_order_acquire); }
240 : :
241 : 0 : const Conf* DdsServerImpl::get_conf() const { return &conf_; }
242 : :
243 : 0 : const AbstractNode* DdsServerImpl::get_abstract_node() const { return this; }
244 : :
245 : 8 : Status::BasePtr DdsServerImpl::get_status(Status::Type type) const {
246 [ + + ]: 8 : if (Status::is_for_writer(type)) {
247 [ + + ]: 3 : if (writer_listener_) {
248 : 1 : return WriterListener::get_status(writer_.get(), type);
249 : : }
250 : 2 : return std::make_shared<Status::Unknown>();
251 : : }
252 : :
253 [ + + ]: 5 : if VUNLIKELY (!reader_) {
254 : 3 : return std::make_shared<Status::Unknown>();
255 : : }
256 : :
257 : 2 : return ReaderListener::get_status(reader_.get(), type);
258 : : }
259 : :
260 : 0 : std::any DdsServerImpl::get_native_handle() const { return subscriber_; }
261 : :
262 : 0 : bool DdsServerImpl::has_clients() const { return read_session_count_.load(std::memory_order_relaxed) > 0; }
263 : :
264 : 28 : bool DdsServerImpl::listen(ReqRespCallback&& callback) {
265 [ - + ]: 28 : if VUNLIKELY (callback_) {
266 : 0 : return false;
267 : : }
268 : :
269 : 28 : callback_ = std::move(callback);
270 : :
271 [ + - ]: 28 : reader_listener_.emplace(this);
272 : :
273 : 56 : reader_ = DdsFactory::create_datareader(kServer, conf_, subscriber_.get(), topic_req_.get(),
274 : 56 : &reader_listener_.value(), is_cdr_type);
275 : :
276 : 28 : return true;
277 : : }
278 : :
279 : 21 : bool DdsServerImpl::reply(uint64_t req_id, const Bytes& resp_data, bool is_sync) {
280 : : (void)is_sync;
281 : :
282 : 21 : bool ret = false;
283 : :
284 [ - + ]: 21 : if (is_cdr_type) {
285 [ # # ]: 0 : std::lock_guard lock(param_mtx_);
286 [ # # ]: 0 : auto iter = cdr_id_map_.find(req_id);
287 : :
288 [ # # ]: 0 : if VUNLIKELY (iter == cdr_id_map_.end()) {
289 [ # # # # ]: 0 : VLOG_E("DdsServer: Cannot find request id.");
290 : 0 : return false;
291 : : }
292 : :
293 [ # # ]: 0 : ret = DdsFactory::write_cdr_data(writer_.get(), resp_data, &iter->second);
294 : :
295 [ # # ]: 0 : if VLIKELY (ret) {
296 [ # # ]: 0 : cdr_id_map_.erase(iter);
297 : : }
298 [ # # ]: 0 : } else {
299 : 21 : ret = DdsFactory::write_data(writer_.get(), resp_data, req_id);
300 : : }
301 : :
302 : 21 : return ret;
303 : : }
304 : :
305 : : } // namespace vlink
|
