Skip to content

Commit a4b4097

Browse files
EugenioColladoEugenioCollado
committed
Implement fast rpc protocol
Signed-off-by: Eugenio Collado <eugeniocollado@eprosima.com>
1 parent 17f24cc commit a4b4097

9 files changed

Lines changed: 346 additions & 132 deletions

File tree

‎ddsenabler/src/cpp/DDSEnabler.cpp‎

Lines changed: 80 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -295,16 +295,26 @@ bool DDSEnabler::send_service_request(
295295
uint64_t& request_id,
296296
participants::RPC_PROTOCOL rpc_protocol)
297297
{
298-
if (rpc_protocol != participants::RPC_PROTOCOL::ROS2)
299-
{
300-
EPROSIMA_LOG_ERROR(DDSENABLER_EXECUTION,
301-
"Failed to send service request to service " << service_name << ": only ROS2 services are currently supported.");
302-
return false;
298+
std::string prefix, suffix;
299+
switch (rpc_protocol)
300+
{
301+
case participants::RPC_PROTOCOL::ROS2:
302+
prefix = participants::ROS_REQUEST_PREFIX;
303+
suffix = participants::ROS_REQUEST_SUFFIX;
304+
break;
305+
case participants::RPC_PROTOCOL::FASTDDS:
306+
prefix = participants::FASTDDS_REQUEST_PREFIX;
307+
suffix = participants::FASTDDS_REQUEST_SUFFIX;
308+
break;
309+
default:
310+
EPROSIMA_LOG_ERROR(DDSENABLER_EXECUTION,
311+
"Failed to send service request to service " << service_name << ": unknown RPC protocol.");
312+
return false;
303313
}
304314

305315
request_id = handler_->get_new_request_id();
306316
if (!enabler_participant_->publish_rpc(
307-
std::string(participants::REQUEST_PREFIX) + service_name + participants::REQUEST_SUFFIX,
317+
prefix + service_name + suffix,
308318
json,
309319
request_id))
310320
return false;
@@ -330,7 +340,28 @@ bool DDSEnabler::send_service_reply(
330340
const std::string& json,
331341
const uint64_t request_id)
332342
{
333-
return enabler_participant_->publish_rpc(std::string(participants::REPLY_PREFIX) + service_name + participants::REPLY_SUFFIX, json, request_id);
343+
RPC_PROTOCOL rpc_protocol = enabler_participant_->get_service_rpc_protocol(service_name);
344+
std::string prefix, suffix;
345+
switch (rpc_protocol)
346+
{
347+
case participants::RPC_PROTOCOL::ROS2:
348+
prefix = participants::ROS_REPLY_PREFIX;
349+
suffix = participants::ROS_REPLY_SUFFIX;
350+
break;
351+
case participants::RPC_PROTOCOL::FASTDDS:
352+
prefix = participants::FASTDDS_REPLY_PREFIX;
353+
suffix = participants::FASTDDS_REPLY_SUFFIX;
354+
break;
355+
default:
356+
EPROSIMA_LOG_ERROR(DDSENABLER_EXECUTION,
357+
"Failed to send service reply to unknown service " << service_name);
358+
return false;
359+
}
360+
361+
return enabler_participant_->publish_rpc(
362+
prefix + service_name + suffix,
363+
json,
364+
request_id);
334365
}
335366

336367
bool DDSEnabler::send_action_goal(
@@ -339,21 +370,15 @@ bool DDSEnabler::send_action_goal(
339370
UUID& action_id,
340371
participants::RPC_PROTOCOL rpc_protocol)
341372
{
342-
if (rpc_protocol != participants::RPC_PROTOCOL::ROS2)
343-
{
344-
EPROSIMA_LOG_ERROR(DDSENABLER_EXECUTION,
345-
"Failed to send action goal to action " << action_name << ": only ROS2 actions are currently supported.");
346-
return false;
347-
}
348-
349373
std::string goal_json = RpcUtils::create_goal_request_msg(json, action_id);
350374
std::string goal_request_topic = action_name + participants::ACTION_GOAL_SUFFIX;
351375
uint64_t goal_request_id = 0;
352376

353377
if (!send_service_request(
354378
goal_request_topic,
355379
goal_json,
356-
goal_request_id))
380+
goal_request_id,
381+
rpc_protocol))
357382
{
358383
EPROSIMA_LOG_ERROR(DDSENABLER_EXECUTION,
359384
"Failed to send action goal request to action " << action_name);
@@ -364,7 +389,8 @@ bool DDSEnabler::send_action_goal(
364389
action_name,
365390
action_id,
366391
goal_request_id,
367-
RpcUtils::ActionType::GOAL))
392+
RpcUtils::ActionType::GOAL,
393+
rpc_protocol))
368394
{
369395
EPROSIMA_LOG_ERROR(DDSENABLER_EXECUTION,
370396
"Failed to store action goal request to action " << action_name);
@@ -558,8 +584,26 @@ bool DDSEnabler::send_action_feedback(
558584
return false;
559585
}
560586

587+
RPC_PROTOCOL protocol = handler_->get_action_rpc_protocol(action_name, goal_id);
588+
589+
std::string prefix;
590+
switch (protocol)
591+
{
592+
case RPC_PROTOCOL::ROS2:
593+
prefix = participants::ROS_TOPIC_PREFIX;
594+
break;
595+
case RPC_PROTOCOL::FASTDDS:
596+
prefix = participants::FASTDDS_TOPIC_PREFIX;
597+
break;
598+
default:
599+
EPROSIMA_LOG_ERROR(DDSENABLER_EXECUTION,
600+
"Failed to send feedback to action " << action_name
601+
<< ": unsupported RPC protocol.");
602+
return false;
603+
}
604+
561605
std::string feedback_json = participants::RpcUtils::create_feedback_msg(json, goal_id);
562-
std::string feedback_topic = participants::TOPIC_PREFIX + std::string(action_name) + participants::ACTION_FEEDBACK_SUFFIX;
606+
std::string feedback_topic = prefix + std::string(action_name) + participants::ACTION_FEEDBACK_SUFFIX;
563607

564608
return enabler_participant_->publish(feedback_topic, feedback_json);
565609
}
@@ -578,8 +622,26 @@ bool DDSEnabler::update_action_status(
578622
return false;
579623
}
580624

625+
RPC_PROTOCOL protocol = handler_->get_action_rpc_protocol(action_name, goal_id);
626+
627+
std::string prefix;
628+
switch (protocol)
629+
{
630+
case RPC_PROTOCOL::ROS2:
631+
prefix = participants::ROS_TOPIC_PREFIX;
632+
break;
633+
case RPC_PROTOCOL::FASTDDS:
634+
prefix = participants::FASTDDS_TOPIC_PREFIX;
635+
break;
636+
default:
637+
EPROSIMA_LOG_ERROR(DDSENABLER_EXECUTION,
638+
"Failed to send status to action " << action_name
639+
<< ": unsupported RPC protocol.");
640+
return false;
641+
}
642+
581643
std::string status_json = participants::RpcUtils::create_status_msg(goal_id, status_code, goal_accepted_stamp);
582-
std::string status_topic = participants::TOPIC_PREFIX + action_name + participants::ACTION_STATUS_SUFFIX;
644+
std::string status_topic = prefix + action_name + participants::ACTION_STATUS_SUFFIX;
583645
return enabler_participant_->publish(status_topic, status_json);
584646
}
585647

‎ddsenabler_participants/include/ddsenabler_participants/Constants.hpp‎

Lines changed: 11 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -29,13 +29,19 @@ constexpr const char* QOS_SERIALIZATION_OWNERSHIP("ownership");
2929
constexpr const char* QOS_SERIALIZATION_KEYED("keyed");
3030

3131
// Topic mangling
32-
constexpr const char* TOPIC_PREFIX("rt/");
32+
constexpr const char* ROS_TOPIC_PREFIX("rt/");
33+
constexpr const char* FASTDDS_TOPIC_PREFIX("");
3334

3435
// Service mangling
35-
constexpr const char* REQUEST_PREFIX("rq/");
36-
constexpr const char* REQUEST_SUFFIX("Request");
37-
constexpr const char* REPLY_PREFIX("rr/");
38-
constexpr const char* REPLY_SUFFIX("Reply");
36+
constexpr const char* ROS_REQUEST_PREFIX("rq/");
37+
constexpr const char* ROS_REQUEST_SUFFIX("Request");
38+
constexpr const char* ROS_REPLY_PREFIX("rr/");
39+
constexpr const char* ROS_REPLY_SUFFIX("Reply");
40+
41+
constexpr const char* FASTDDS_REQUEST_PREFIX("");
42+
constexpr const char* FASTDDS_REQUEST_SUFFIX("_Request");
43+
constexpr const char* FASTDDS_REPLY_PREFIX("");
44+
constexpr const char* FASTDDS_REPLY_SUFFIX("_Reply");
3945

4046
// Action mangling
4147
constexpr const char* ACTION_GOAL_SUFFIX("send_goal");

‎ddsenabler_participants/include/ddsenabler_participants/EnablerParticipant.hpp‎

Lines changed: 13 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -98,6 +98,10 @@ class EnablerParticipant : public ddspipe::participants::SchemaParticipant
9898
bool revoke_service(
9999
const std::string& service_name);
100100

101+
DDSENABLER_PARTICIPANTS_DllAPI
102+
RPC_PROTOCOL get_service_rpc_protocol(
103+
const std::string& service_name);
104+
101105
DDSENABLER_PARTICIPANTS_DllAPI
102106
bool announce_action(
103107
const std::string& action_name,
@@ -114,10 +118,12 @@ class EnablerParticipant : public ddspipe::participants::SchemaParticipant
114118
ddspipe::core::types::DdsTopic& topic);
115119

116120
bool query_service_nts_(
117-
std::shared_ptr<ServiceDiscovered> service);
121+
std::shared_ptr<ServiceDiscovered> service,
122+
RPC_PROTOCOL rpc_protocol);
118123

119124
bool query_action_nts_(
120125
ActionDiscovered& action,
126+
RPC_PROTOCOL rpc_protocol,
121127
std::unique_lock<std::mutex>& lck);
122128

123129
bool create_topic_writer_nts_(
@@ -137,7 +143,8 @@ class EnablerParticipant : public ddspipe::participants::SchemaParticipant
137143

138144
bool fullfill_service_type_nts_(
139145
const ServiceInfo& service_info,
140-
std::shared_ptr<ServiceDiscovered> service);
146+
std::shared_ptr<ServiceDiscovered> service,
147+
RPC_PROTOCOL rpc_protocol);
141148

142149
std::shared_ptr<ddspipe::core::IReader> lookup_reader_nts_(
143150
const std::string& topic_name,
@@ -149,12 +156,14 @@ class EnablerParticipant : public ddspipe::participants::SchemaParticipant
149156
bool service_discovered_nts_(
150157
const std::string& service_name,
151158
const ddspipe::core::types::DdsTopic& topic,
152-
RpcUtils::RpcType rpc_type);
159+
RpcUtils::RpcType rpc_type,
160+
RPC_PROTOCOL rpc_protocol);
153161

154162
bool action_discovered_nts_(
155163
const std::string& action_name,
156164
const ddspipe::core::types::DdsTopic& topic,
157-
RpcUtils::RpcType rpc_type);
165+
RpcUtils::RpcType rpc_type,
166+
RPC_PROTOCOL rpc_protocol);
158167

159168
bool revoke_service_nts_(
160169
const std::string& service_name);

‎ddsenabler_participants/include/ddsenabler_participants/Handler.hpp‎

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -191,14 +191,16 @@ class Handler : public ddspipe::participants::ISchemaHandler
191191
* @param [in] action_id UUID of the action.
192192
* @param [in] request_id Request ID of the action request.
193193
* @param [in] action_type Type of the action (GOAL, RESULT, CANCEL).
194+
* @param [in] rpc_protocol Protocol of the action, only used to create the ActionRequestInfo if action_type is GOAL.
194195
* @return \c true if the action request was successfully stored, \c false otherwise.
195196
*/
196197
DDSENABLER_PARTICIPANTS_DllAPI
197198
bool store_action_request(
198199
const std::string& action_name,
199200
const UUID& action_id,
200201
const uint64_t request_id,
201-
const RpcUtils::ActionType action_type);
202+
const RpcUtils::ActionType action_type,
203+
const RPC_PROTOCOL rpc_protocol = RPC_PROTOCOL::ROS2);
202204

203205
/**
204206
* @brief Send the reply containing the result of an action or store it for a later reply.
@@ -246,6 +248,18 @@ class Handler : public ddspipe::participants::ISchemaHandler
246248
const UUID& action_id,
247249
std::chrono::system_clock::time_point* goal_accepted_stamp = nullptr);
248250

251+
/**
252+
* @brief Get the RPC protocol of an action represented by its UUID.
253+
*
254+
* @param [in] action_name Name of the action.
255+
* @param [in] action_id UUID of the action to be checked.
256+
* @return The RPC protocol of the action, or RPC_PROTOCOL::UNKNOWN if the action is not found.
257+
*/
258+
DDSENABLER_PARTICIPANTS_DllAPI
259+
RPC_PROTOCOL get_action_rpc_protocol(
260+
const std::string& action_name,
261+
const UUID& action_id);
262+
249263
/**
250264
* @brief Set the data notification callback.
251265
*

‎ddsenabler_participants/include/ddsenabler_participants/RpcStructs.hpp‎

Lines changed: 24 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -41,9 +41,11 @@ struct ActionRequestInfo
4141
ActionRequestInfo(
4242
const std::string& _action_name,
4343
RpcUtils::ActionType action_type,
44-
uint64_t request_id)
44+
uint64_t request_id,
45+
RPC_PROTOCOL rpc_protocol)
4546
: action_name(_action_name)
4647
, goal_accepted_stamp(std::chrono::system_clock::now())
48+
, rpc_protocol(rpc_protocol)
4749
{
4850
set_request(request_id, action_type);
4951
}
@@ -108,7 +110,13 @@ struct ActionRequestInfo
108110
return final_status_received && result_received;
109111
}
110112

113+
RPC_PROTOCOL get_rpc_protocol() const
114+
{
115+
return rpc_protocol;
116+
}
117+
111118
std::string action_name;
119+
RPC_PROTOCOL rpc_protocol;
112120
uint64_t goal_request_id = 0;
113121
uint64_t result_request_id = 0;
114122
std::chrono::system_clock::time_point goal_accepted_stamp;
@@ -120,12 +128,16 @@ struct ActionRequestInfo
120128
struct ServiceDiscovered
121129
{
122130

123-
ServiceDiscovered(const std::string& service_name)
131+
ServiceDiscovered(
132+
const std::string& service_name,
133+
RPC_PROTOCOL rpc_protocol)
124134
: service_name(service_name)
135+
, rpc_protocol(rpc_protocol)
125136
{
126137
}
127138

128139
std::string service_name;
140+
RPC_PROTOCOL rpc_protocol{RPC_PROTOCOL::UNKNOWN};
129141

130142
ddspipe::core::types::DdsTopic topic_request;
131143
bool request_discovered{false};
@@ -221,16 +233,25 @@ struct ServiceDiscovered
221233
}
222234
return false;
223235
}
236+
237+
RPC_PROTOCOL get_rpc_protocol() const
238+
{
239+
return rpc_protocol;
240+
}
224241
};
225242

226243
struct ActionDiscovered
227244
{
228-
ActionDiscovered(const std::string& action_name)
245+
ActionDiscovered(
246+
const std::string& action_name,
247+
RPC_PROTOCOL rpc_protocol)
229248
: action_name(action_name)
249+
, rpc_protocol(rpc_protocol)
230250
{
231251
}
232252

233253
std::string action_name;
254+
RPC_PROTOCOL rpc_protocol{RPC_PROTOCOL::UNKNOWN};
234255
std::weak_ptr<ServiceDiscovered> goal;
235256
std::weak_ptr<ServiceDiscovered> result;
236257
std::weak_ptr<ServiceDiscovered> cancel;

‎ddsenabler_participants/include/ddsenabler_participants/RpcUtils.hpp‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -88,7 +88,11 @@ enum ActionType
8888
* @param [in] topic_name Topic name to extract the service/action name from
8989
* @return Extracted service name
9090
*/
91-
RpcType get_rpc_name(const std::string& topic_name, std::string& rpc_name);
91+
RpcType get_rpc_name(const std::string& topic_name, std::string& rpc_name, RPC_PROTOCOL& rpc_protocol);
92+
93+
RPC_PROTOCOL detect_rpc_protocol(const std::string& topic_name);
94+
95+
RpcType remove_prefix_suffix(const std::string& topic_name, std::string& rpc_name, RPC_PROTOCOL rpc_protocol);
9296

9397
RpcType get_service_name(const std::string& topic_name, std::string& service_name);
9498

0 commit comments

Comments
 (0)