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