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_server_impl.hpp"
25 : :
26 : : #include <memory>
27 : : #include <utility>
28 : :
29 : : #include "./base/message_loop.h"
30 : :
31 : : namespace vlink {
32 : :
33 : : // WriterListener
34 : 21 : DdscServerImpl::WriterListener::WriterListener(NodeImpl* impl) : DdscWriterListener(impl) {}
35 : :
36 : : // ReaderListener
37 : 25 : DdscServerImpl::ReaderListener::ReaderListener(NodeImpl* impl) : DdscReaderListener(impl) {}
38 : :
39 : 44 : void DdscServerImpl::ReaderListener::on_subscription_matched(NodeImpl* impl,
40 : : const dds_subscription_matched_status_t& status) {
41 : 44 : auto* instance = static_cast<DdscServerImpl*>(impl);
42 : :
43 : 44 : instance->read_session_count_.store(status.current_count, std::memory_order_relaxed);
44 : :
45 : 44 : DdscReaderListener::on_subscription_matched(impl, status);
46 : 44 : }
47 : :
48 : 49 : void DdscServerImpl::ReaderListener::on_data_available(NodeImpl* impl, dds_entity_t reader) {
49 : 49 : auto* instance = static_cast<DdscServerImpl*>(impl);
50 : 49 : auto* message_loop = instance->get_message_loop();
51 : :
52 [ + + ]: 49 : if VUNLIKELY (instance->has_suspend.load(std::memory_order_acquire)) {
53 : 1 : DdscFactory::ReadMessage msg;
54 : :
55 [ + - + + ]: 2 : while (DdscFactory::take_data(reader, msg)) {
56 [ + - ]: 1 : DdscFactory::release_data(reader, msg);
57 : :
58 [ - + ]: 1 : if VUNLIKELY (instance->quit_flag_.load(std::memory_order_acquire)) {
59 : 0 : break;
60 : : }
61 : : }
62 : :
63 : 1 : return;
64 : 1 : }
65 : :
66 [ - + ]: 48 : if VUNLIKELY (!instance->callback_) {
67 : 0 : return;
68 : : }
69 : :
70 [ + + ]: 48 : if (message_loop) {
71 [ + - + - ]: 2 : message_loop->post_task([instance, reader]() {
72 [ - + ]: 2 : if VUNLIKELY (!instance->get_message_loop()) {
73 : 0 : return;
74 : : }
75 : :
76 : 2 : instance->process_message(reader);
77 : : });
78 : : } else {
79 : 46 : instance->process_message(reader);
80 : : }
81 : : }
82 : :
83 : : // DdscServerImpl
84 [ + - + - ]: 34 : DdscServerImpl::DdscServerImpl(const DdscConf& conf) : conf_(conf) {}
85 : :
86 : 48 : void DdscServerImpl::process_message(dds_entity_t reader) {
87 : 48 : DdscFactory::ReadMessage msg;
88 : :
89 [ + - + + ]: 96 : while (DdscFactory::take_data(reader, msg)) {
90 [ - + ]: 48 : if VUNLIKELY (quit_flag_.load(std::memory_order_acquire)) {
91 [ # # ]: 0 : DdscFactory::release_data(reader, msg);
92 : 0 : break;
93 : : }
94 : :
95 [ + + ]: 48 : if VUNLIKELY (!msg.info.valid_data) {
96 [ + - ]: 19 : DdscFactory::release_data(reader, msg);
97 : 19 : continue;
98 : : }
99 : :
100 [ + + ]: 29 : if (writer_) {
101 : 24 : Bytes resp_data;
102 [ + - ]: 24 : callback_(msg.id, msg.bytes, &resp_data);
103 : 24 : } else {
104 [ + - ]: 5 : callback_(msg.id, msg.bytes, nullptr);
105 : : }
106 : :
107 [ + - ]: 29 : DdscFactory::release_data(reader, msg);
108 : : }
109 : 48 : }
110 : :
111 : 29 : void DdscServerImpl::init() {
112 [ + - ]: 29 : participant_ = DdscFactory::create_participant(kServer | kClient, conf_, get_all_properties());
113 : :
114 [ + + ]: 29 : if (is_resp_type) {
115 : 21 : std::tie(topic_req_, topic_resp_) = DdscFactory::create_method_topic(kServer | kClient, conf_, participant_.get());
116 : :
117 : 21 : publisher_ = DdscFactory::create_publisher(kServer, conf_, participant_.get());
118 : :
119 [ + - ]: 21 : writer_listener_.emplace(this);
120 : :
121 : 42 : writer_ = DdscFactory::create_datawriter(kServer, conf_, publisher_.get(), topic_resp_.get(),
122 : 42 : writer_listener_->get_ptr());
123 : : } else {
124 [ + - + - ]: 8 : topic_req_ = DdscFactory::create_topic(kServer | kClient, conf_, participant_.get());
125 : : }
126 : :
127 : 29 : subscriber_ = DdscFactory::create_subscriber(kServer, conf_, participant_.get());
128 : :
129 [ + - - + : 29 : if VUNLIKELY (!participant_ || !topic_req_) {
- + ]
130 [ # # # # ]: 0 : VLOG_E("DdscServerImpl::init(): participant/topic creation failed; server left uninitialised.");
131 : :
132 : 0 : return;
133 : : }
134 : :
135 : 29 : quit_flag_.store(false, std::memory_order_release);
136 : : }
137 : :
138 : 29 : void DdscServerImpl::deinit() {
139 : 29 : quit_flag_.store(true, std::memory_order_release);
140 : :
141 : 29 : detach();
142 : :
143 : 29 : reader_.reset();
144 : 29 : writer_.reset();
145 : 29 : reader_listener_.reset();
146 : 29 : writer_listener_.reset();
147 : 29 : subscriber_.reset();
148 : 29 : publisher_.reset();
149 : 29 : topic_resp_.reset();
150 : 29 : topic_req_.reset();
151 : 29 : participant_.reset();
152 : 29 : }
153 : :
154 : 3 : bool DdscServerImpl::suspend() {
155 : 3 : has_suspend.store(true, std::memory_order_release);
156 : :
157 : 3 : return true;
158 : : }
159 : :
160 : 3 : bool DdscServerImpl::resume() {
161 : 3 : has_suspend.store(false, std::memory_order_release);
162 : :
163 : 3 : return true;
164 : : }
165 : :
166 : 6 : bool DdscServerImpl::is_suspend() const { return has_suspend.load(std::memory_order_acquire); }
167 : :
168 : 0 : const Conf* DdscServerImpl::get_conf() const { return &conf_; }
169 : :
170 : 0 : const AbstractNode* DdscServerImpl::get_abstract_node() const { return this; }
171 : :
172 : 8 : Status::BasePtr DdscServerImpl::get_status(Status::Type type) const {
173 [ + + ]: 8 : if (Status::is_for_writer(type)) {
174 [ + + ]: 3 : if (writer_listener_) {
175 : 1 : return WriterListener::get_status(writer_->entity, type);
176 : : }
177 : :
178 : 2 : return std::make_shared<Status::Unknown>();
179 : : }
180 : :
181 [ + + ]: 5 : if VUNLIKELY (!reader_) {
182 : 3 : return std::make_shared<Status::Unknown>();
183 : : }
184 : :
185 : 2 : return ReaderListener::get_status(reader_->entity, type);
186 : : }
187 : :
188 : 0 : std::any DdscServerImpl::get_native_handle() const { return subscriber_; }
189 : :
190 : 0 : bool DdscServerImpl::has_clients() const { return read_session_count_.load(std::memory_order_relaxed) > 0; }
191 : :
192 : 25 : bool DdscServerImpl::listen(ReqRespCallback&& callback) {
193 [ - + ]: 25 : if VUNLIKELY (callback_) {
194 : 0 : return false;
195 : : }
196 : :
197 : 25 : callback_ = std::move(callback);
198 : :
199 [ + - ]: 25 : reader_listener_.emplace(this);
200 : :
201 : : reader_ =
202 : 25 : DdscFactory::create_datareader(kServer, conf_, subscriber_.get(), topic_req_.get(), reader_listener_->get_ptr());
203 : :
204 : 25 : return true;
205 : : }
206 : :
207 : 20 : bool DdscServerImpl::reply(uint64_t req_id, const Bytes& resp_data, bool is_sync) {
208 : : (void)is_sync;
209 : :
210 : 20 : bool ret = DdscFactory::write_data(writer_->entity, resp_data, req_id);
211 : :
212 : 20 : return ret;
213 : : }
214 : :
215 : : } // namespace vlink
|
