Skip to content

Commit bf55d52

Browse files
rsanchez15rsanchez15
authored
Update RPC example types and code (#5894)
Signed-off-by: Raul Sanchez-Mateos <[email protected]>
1 parent 8f63449 commit bf55d52

9 files changed

+3233
-96
lines changed

‎examples/cpp/rpc/ServerApp.cpp‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -136,7 +136,7 @@ void ServerApp::create_server(
136136
}
137137

138138
calculator_example::detail::Calculator_representation_limits_Out ServerApp::ServerImpl::representation_limits(
139-
const CalculatorServer_ClientContext& info)
139+
const eprosima::fastdds::dds::rpc::RpcRequest& info)
140140
{
141141
static_cast<void>(info);
142142
calculator_example::detail::Calculator_representation_limits_Out limits;
@@ -146,7 +146,7 @@ calculator_example::detail::Calculator_representation_limits_Out ServerApp::Serv
146146
}
147147

148148
int32_t ServerApp::ServerImpl::addition(
149-
const CalculatorServer_ClientContext& info,
149+
const eprosima::fastdds::dds::rpc::RpcRequest& info,
150150
/*in*/ int32_t value1,
151151
/*in*/ int32_t value2)
152152
{
@@ -166,7 +166,7 @@ int32_t ServerApp::ServerImpl::addition(
166166
}
167167

168168
int32_t ServerApp::ServerImpl::subtraction(
169-
const CalculatorServer_ClientContext& info,
169+
const eprosima::fastdds::dds::rpc::RpcRequest& info,
170170
/*in*/ int32_t value1,
171171
/*in*/ int32_t value2)
172172
{
@@ -186,7 +186,7 @@ int32_t ServerApp::ServerImpl::subtraction(
186186
}
187187

188188
void ServerApp::ServerImpl::fibonacci_seq(
189-
const CalculatorServer_ClientContext& info,
189+
const eprosima::fastdds::dds::rpc::RpcRequest& info,
190190
/*in*/ uint32_t n_results,
191191
/*result*/ eprosima::fastdds::dds::rpc::RpcServerWriter<int32_t>& result_writer)
192192
{
@@ -213,7 +213,7 @@ void ServerApp::ServerImpl::fibonacci_seq(
213213
}
214214

215215
int32_t ServerApp::ServerImpl::sum_all(
216-
const CalculatorServer_ClientContext& info,
216+
const eprosima::fastdds::dds::rpc::RpcRequest& info,
217217
/*in*/ eprosima::fastdds::dds::rpc::RpcServerReader<int32_t>& value)
218218
{
219219
static_cast<void>(info);
@@ -246,7 +246,7 @@ int32_t ServerApp::ServerImpl::sum_all(
246246
}
247247

248248
void ServerApp::ServerImpl::accumulator(
249-
const CalculatorServer_ClientContext& info,
249+
const eprosima::fastdds::dds::rpc::RpcRequest& info,
250250
/*in*/ eprosima::fastdds::dds::rpc::RpcServerReader<int32_t>& value,
251251
/*result*/ eprosima::fastdds::dds::rpc::RpcServerWriter<int32_t>& result_writer)
252252
{
@@ -279,7 +279,7 @@ void ServerApp::ServerImpl::accumulator(
279279
}
280280

281281
void ServerApp::ServerImpl::filter(
282-
const CalculatorServer_ClientContext& info,
282+
const eprosima::fastdds::dds::rpc::RpcRequest& info,
283283
/*in*/ eprosima::fastdds::dds::rpc::RpcServerReader<int32_t>& value,
284284
/*in*/ calculator_example::FilterKind filter_kind,
285285
/*result*/ eprosima::fastdds::dds::rpc::RpcServerWriter<int32_t>& result_writer)

‎examples/cpp/rpc/ServerApp.hpp‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -80,42 +80,42 @@ class ServerApp : public Application
8080
explicit ServerImpl() = default;
8181

8282
calculator_example::detail::Calculator_representation_limits_Out representation_limits(
83-
const calculator_example::CalculatorServer_ClientContext& info) override;
83+
const eprosima::fastdds::dds::rpc::RpcRequest& info) override;
8484

8585
int32_t addition(
86-
const calculator_example::CalculatorServer_ClientContext& info,
86+
const eprosima::fastdds::dds::rpc::RpcRequest& info,
8787
/*in*/ int32_t value1,
8888
/*in*/ int32_t value2) override;
8989

9090
int32_t subtraction(
91-
const calculator_example::CalculatorServer_ClientContext& info,
91+
const eprosima::fastdds::dds::rpc::RpcRequest& info,
9292
/*in*/ int32_t value1,
9393
/*in*/ int32_t value2) override;
9494

9595
void fibonacci_seq(
96-
const calculator_example::CalculatorServer_ClientContext& info,
96+
const eprosima::fastdds::dds::rpc::RpcRequest& info,
9797
/*in*/ uint32_t n_results,
9898
/*result*/ eprosima::fastdds::dds::rpc::RpcServerWriter<int32_t>& result_writer) override;
9999

100100
int32_t sum_all(
101-
const calculator_example::CalculatorServer_ClientContext& info,
101+
const eprosima::fastdds::dds::rpc::RpcRequest& info,
102102
/*in*/ eprosima::fastdds::dds::rpc::RpcServerReader<int32_t>& value) override;
103103

104104
void accumulator(
105-
const calculator_example::CalculatorServer_ClientContext& info,
105+
const eprosima::fastdds::dds::rpc::RpcRequest& info,
106106
/*in*/ eprosima::fastdds::dds::rpc::RpcServerReader<int32_t>& value,
107107
/*result*/ eprosima::fastdds::dds::rpc::RpcServerWriter<int32_t>& result_writer) override;
108108

109109
void filter(
110-
const calculator_example::CalculatorServer_ClientContext& info,
110+
const eprosima::fastdds::dds::rpc::RpcRequest& info,
111111
/*in*/ eprosima::fastdds::dds::rpc::RpcServerReader<int32_t>& value,
112112
/*in*/ calculator_example::FilterKind filter_kind,
113113
/*result*/ eprosima::fastdds::dds::rpc::RpcServerWriter<int32_t>& result_writer) override;
114114

115115
};
116116

117117
std::shared_ptr<ServerImpl> server_impl_;
118-
std::shared_ptr<calculator_example::CalculatorServer> server_;
118+
std::shared_ptr<eprosima::fastdds::dds::rpc::RpcServer> server_;
119119
dds::DomainParticipant* participant_;
120120
size_t thread_pool_size_;
121121
std::atomic<bool> stop_;

‎examples/cpp/rpc/types/calculatorClient.cxx‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -205,6 +205,7 @@ class CalculatorClient : public Calculator
205205
std::shared_ptr<T> result,
206206
std::promise<TResult>& promise)
207207
{
208+
result->info.related_sample_identity.writer_guid(requester_->get_requester_reader()->guid());
208209
if (fdds::RETCODE_OK == requester_->send_request((void*)&request, result->info))
209210
{
210211
std::lock_guard<std::mutex> _ (mtx_);
@@ -224,6 +225,7 @@ class CalculatorClient : public Calculator
224225
const RequestType& request,
225226
std::shared_ptr<T> result)
226227
{
228+
result->info.related_sample_identity.writer_guid(requester_->get_requester_reader()->guid());
227229
if (fdds::RETCODE_OK == requester_->send_request((void*)&request, result->info))
228230
{
229231
std::lock_guard<std::mutex> _ (mtx_);

‎examples/cpp/rpc/types/calculatorPubSubTypes.cxx‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -175,6 +175,11 @@ namespace calculator_example {
175175
delete pData;
176176
}
177177

178+
eProsima_user_DllExport void register_type_object_representation() override
179+
{
180+
register_Calculator_Request_type_identifier(type_identifiers_);
181+
}
182+
178183
};
179184

180185
class Calculator_ReplyPubSubType : public eprosima::fastdds::dds::TopicDataType
@@ -318,6 +323,11 @@ namespace calculator_example {
318323
delete pData;
319324
}
320325

326+
eProsima_user_DllExport void register_type_object_representation() override
327+
{
328+
register_Calculator_Reply_type_identifier(type_identifiers_);
329+
}
330+
321331
};
322332

323333
eprosima::fastdds::dds::rpc::ServiceTypeSupport create_Calculator_service_type_support()

‎examples/cpp/rpc/types/calculatorServer.cxx‎

Lines changed: 104 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -63,22 +63,38 @@ namespace frpc = eprosima::fastdds::dds::rpc;
6363
namespace frtps = eprosima::fastdds::rtps;
6464

6565
class CalculatorServerLogic
66-
: public CalculatorServer
66+
: public frpc::RpcServer
67+
, public std::enable_shared_from_this<CalculatorServerLogic>
6768
{
6869
using RequestType = Calculator_Request;
6970
using ReplyType = Calculator_Reply;
7071

7172
public:
7273

7374
CalculatorServerLogic(
74-
eprosima::fastdds::dds::DomainParticipant& part,
75+
fdds::DomainParticipant& part,
7576
const char* service_name,
76-
const eprosima::fastdds::dds::ReplierQos& qos,
77+
const fdds::ReplierQos& qos,
7778
size_t thread_pool_size,
7879
std::shared_ptr<CalculatorServer_IServerImplementation> implementation)
79-
: CalculatorServer()
80+
: CalculatorServerLogic(
81+
part,
82+
service_name,
83+
qos,
84+
std::make_shared<ThreadPool>(*this, thread_pool_size),
85+
std::move(implementation))
86+
{
87+
}
88+
89+
CalculatorServerLogic(
90+
fdds::DomainParticipant& part,
91+
const char* service_name,
92+
const fdds::ReplierQos& qos,
93+
std::shared_ptr<frpc::RpcServerSchedulingStrategy> scheduler,
94+
std::shared_ptr<CalculatorServer_IServerImplementation> implementation)
95+
: frpc::RpcServer()
8096
, participant_(part)
81-
, thread_pool_(*this, thread_pool_size)
97+
, request_scheduler_(scheduler)
8298
, implementation_(std::move(implementation))
8399
{
84100
// Register the service type support
@@ -106,8 +122,6 @@ class CalculatorServerLogic
106122

107123
~CalculatorServerLogic() override
108124
{
109-
stop();
110-
111125
if (nullptr != replier_)
112126
{
113127
participant_.delete_service_replier(service_->get_service_name(), replier_);
@@ -174,7 +188,21 @@ class CalculatorServerLogic
174188
}
175189

176190
// Wait for all threads to finish
177-
thread_pool_.stop();
191+
request_scheduler_->server_stopped(shared_from_this());
192+
}
193+
194+
void execute_request(
195+
const std::shared_ptr<frpc::RpcRequest>& request) override
196+
{
197+
auto ctx = std::dynamic_pointer_cast<RequestContext>(request);
198+
if (ctx)
199+
{
200+
execute_request(ctx);
201+
}
202+
else
203+
{
204+
throw std::runtime_error("Invalid request context type");
205+
}
178206
}
179207

180208
private:
@@ -724,7 +752,7 @@ class CalculatorServerLogic
724752

725753
//} operation filter
726754

727-
struct RequestContext : CalculatorServer_ClientContext
755+
struct RequestContext : frpc::RpcRequest
728756
{
729757
RequestType request;
730758
frpc::RequestInfo info;
@@ -999,6 +1027,7 @@ class CalculatorServerLogic
9991027
};
10001028

10011029
struct ThreadPool
1030+
: public frpc::RpcServerSchedulingStrategy
10021031
{
10031032
ThreadPool(
10041033
CalculatorServerLogic& server,
@@ -1015,7 +1044,7 @@ class CalculatorServerLogic
10151044
{
10161045
while (!finished_)
10171046
{
1018-
std::shared_ptr<RequestContext> req;
1047+
std::shared_ptr<frpc::RpcRequest> req;
10191048
{
10201049
std::unique_lock<std::mutex> lock(mtx_);
10211050
cv_.wait(lock, [this]()
@@ -1041,9 +1070,12 @@ class CalculatorServerLogic
10411070
}
10421071
}
10431072

1044-
void new_request(
1045-
const std::shared_ptr<RequestContext>& req)
1073+
void schedule_request(
1074+
const std::shared_ptr<frpc::RpcRequest>& req,
1075+
const std::shared_ptr<frpc::RpcServer>& server) override
10461076
{
1077+
static_cast<void>(server);
1078+
10471079
std::lock_guard<std::mutex> lock(mtx_);
10481080
if (!finished_)
10491081
{
@@ -1052,8 +1084,11 @@ class CalculatorServerLogic
10521084
}
10531085
}
10541086

1055-
void stop()
1087+
void server_stopped(
1088+
const std::shared_ptr<frpc::RpcServer>& server) override
10561089
{
1090+
static_cast<void>(server);
1091+
10571092
// Notify all threads in the pool to stop
10581093
{
10591094
std::lock_guard<std::mutex> lock(mtx_);
@@ -1075,7 +1110,7 @@ class CalculatorServerLogic
10751110
CalculatorServerLogic& server_;
10761111
std::mutex mtx_;
10771112
std::condition_variable cv_;
1078-
std::queue<std::shared_ptr<RequestContext>> requests_;
1113+
std::queue<std::shared_ptr<frpc::RpcRequest>> requests_;
10791114
bool finished_{ false };
10801115
std::vector<std::thread> threads_;
10811116
};
@@ -1107,7 +1142,7 @@ class CalculatorServerLogic
11071142
processing_requests_[id] = ctx;
11081143
}
11091144

1110-
thread_pool_.new_request(ctx);
1145+
request_scheduler_->schedule_request(ctx, shared_from_this());
11111146
}
11121147

11131148
void execute_request(
@@ -1312,22 +1347,73 @@ class CalculatorServerLogic
13121347
fdds::GuardCondition finish_condition_;
13131348
std::mutex mtx_;
13141349
std::map<frtps::SampleIdentity, std::shared_ptr<RequestContext>> processing_requests_;
1315-
ThreadPool thread_pool_;
1350+
std::shared_ptr<frpc::RpcServerSchedulingStrategy> request_scheduler_;
13161351
std::shared_ptr<CalculatorServer_IServerImplementation> implementation_;
13171352

13181353
};
13191354

1355+
struct CalculatorServerProxy
1356+
: public frpc::RpcServer
1357+
{
1358+
CalculatorServerProxy(
1359+
std::shared_ptr<frpc::RpcServer> impl)
1360+
: impl_(std::move(impl))
1361+
{
1362+
}
1363+
1364+
~CalculatorServerProxy() override
1365+
{
1366+
if (impl_)
1367+
{
1368+
impl_->stop();
1369+
}
1370+
}
1371+
1372+
void run() override
1373+
{
1374+
impl_->run();
1375+
}
1376+
1377+
void stop() override
1378+
{
1379+
impl_->stop();
1380+
}
1381+
1382+
void execute_request(
1383+
const std::shared_ptr<frpc::RpcRequest>& request) override
1384+
{
1385+
impl_->execute_request(request);
1386+
}
1387+
1388+
private:
1389+
1390+
std::shared_ptr<frpc::RpcServer> impl_;
1391+
};
1392+
13201393
} // namespace detail
13211394

1322-
std::shared_ptr<CalculatorServer> create_CalculatorServer(
1395+
std::shared_ptr<eprosima::fastdds::dds::rpc::RpcServer> create_CalculatorServer(
13231396
eprosima::fastdds::dds::DomainParticipant& part,
13241397
const char* service_name,
13251398
const eprosima::fastdds::dds::ReplierQos& qos,
13261399
size_t thread_pool_size,
13271400
std::shared_ptr<CalculatorServer_IServerImplementation> implementation)
13281401
{
1329-
return std::make_shared<detail::CalculatorServerLogic>(
1402+
auto ptr = std::make_shared<detail::CalculatorServerLogic>(
13301403
part, service_name, qos, thread_pool_size, implementation);
1404+
return std::make_shared<detail::CalculatorServerProxy>(ptr);
1405+
}
1406+
1407+
std::shared_ptr<eprosima::fastdds::dds::rpc::RpcServer> create_CalculatorServer(
1408+
eprosima::fastdds::dds::DomainParticipant& part,
1409+
const char* service_name,
1410+
const eprosima::fastdds::dds::ReplierQos& qos,
1411+
std::shared_ptr<eprosima::fastdds::dds::rpc::RpcServerSchedulingStrategy> scheduler,
1412+
std::shared_ptr<CalculatorServer_IServerImplementation> implementation)
1413+
{
1414+
auto ptr = std::make_shared<detail::CalculatorServerLogic>(
1415+
part, service_name, qos, scheduler, implementation);
1416+
return std::make_shared<detail::CalculatorServerProxy>(ptr);
13311417
}
13321418

13331419
//} interface Calculator

0 commit comments

Comments
 (0)