diff --git a/cyber/service_discovery/communication/participant_listener.h b/cyber/service_discovery/communication/participant_listener.h index 0509094..739e4f1 100644 --- a/cyber/service_discovery/communication/participant_listener.h +++ b/cyber/service_discovery/communication/participant_listener.h @@ -29,7 +29,7 @@ namespace cyber { namespace service_discovery { class ParticipantListener - : public eprosima::fastdds::dds::DomainParticipantImpl { + : public eprosima::fastdds::rtps::RTPSParticipantListener { public: using ChangeFunc = std::function; diff --git a/cyber/service_discovery/communication/subscriber_listener.cc b/cyber/service_discovery/communication/subscriber_listener.cc index 9c3c122..a9561be 100644 --- a/cyber/service_discovery/communication/subscriber_listener.cc +++ b/cyber/service_discovery/communication/subscriber_listener.cc @@ -32,25 +32,18 @@ SubscriberListener::~SubscriberListener() { callback_ = nullptr; } -void SubscriberListener::onNewDataMessage(eprosima::fastdds::dds::Subscriber* sub) { +void SubscriberListener::on_data_available(eprosima::fastdds::dds::DataReader* reader) { RETURN_IF_NULL(callback_); std::lock_guard lock(mutex_); - eprosima::fastrtps::SampleInfo_t m_info; + eprosima::fastdds::dds::SampleInfo m_info; cyber::transport::UnderlayMessage m; - RETURN_IF(!sub->takeNextData(reinterpret_cast(&m), &m_info)); - RETURN_IF(m_info.sampleKind != eprosima::fastdds::rtps::ALIVE); + RETURN_IF( + reader->take_next_sample(reinterpret_cast(&m), &m_info) != eprosima::fastdds::dds::RETCODE_OK); + RETURN_IF(m_info.instance_state != eprosima::fastdds::dds::ALIVE_INSTANCE_STATE); callback_(m.data()); } - -void SubscriberListener::onSubscriptionMatched( - eprosima::fastdds::dds::Subscriber* sub, - eprosima::fastdds::rtps::MatchingInfo& info) { - (void)sub; - (void)info; -} - } // namespace service_discovery } // namespace cyber } // namespace apollo diff --git a/cyber/service_discovery/communication/subscriber_listener.h b/cyber/service_discovery/communication/subscriber_listener.h index 935a1de..a0d4b4a 100644 --- a/cyber/service_discovery/communication/subscriber_listener.h +++ b/cyber/service_discovery/communication/subscriber_listener.h @@ -21,6 +21,9 @@ #include #include +#include "fastdds/dds/subscriber/SampleInfo.hpp" +#include "fastdds/dds/subscriber/DataReader.hpp" +#include "fastdds/dds/subscriber/DataReaderListener.hpp" #include "fastdds/dds/subscriber/Subscriber.hpp" #include "fastdds/dds/subscriber/SubscriberListener.hpp" #include "fastdds/rtps/common/MatchingInfo.hpp" @@ -29,17 +32,14 @@ namespace apollo { namespace cyber { namespace service_discovery { -class SubscriberListener : public eprosima::fastdds::dds::SubscriberListener { +class SubscriberListener : public eprosima::fastdds::dds::DataReaderListener { public: using NewMsgCallback = std::function; explicit SubscriberListener(const NewMsgCallback& callback); virtual ~SubscriberListener(); - void onNewDataMessage(eprosima::fastdds::dds::Subscriber* sub); - void onSubscriptionMatched( - eprosima::fastdds::dds::Subscriber* sub, - eprosima::fastdds::rtps::MatchingInfo& info); // NOLINT + virtual void on_data_available(eprosima::fastdds::dds::DataReader* reader) override; private: NewMsgCallback callback_; diff --git a/cyber/service_discovery/specific_manager/manager.cc b/cyber/service_discovery/specific_manager/manager.cc index 353ef73..9a0954a 100644 --- a/cyber/service_discovery/specific_manager/manager.cc +++ b/cyber/service_discovery/specific_manager/manager.cc @@ -47,10 +47,11 @@ Manager::Manager() Manager::~Manager() { Shutdown(); } -bool Manager::StartDiscovery(RtpsParticipant* participant) { +bool Manager::StartDiscovery(eprosima::fastdds::dds::DomainParticipant* participant) { if (participant == nullptr) { return false; } + participant_ = participant; if (is_discovery_started_.exchange(true)) { return true; } @@ -70,13 +71,17 @@ void Manager::StopDiscovery() { { std::lock_guard lg(lock_); if (publisher_ != nullptr) { - eprosima::fastrtps::Domain::removePublisher(publisher_); + if (participant_) { + participant_->delete_publisher(publisher_); + } publisher_ = nullptr; } } if (subscriber_ != nullptr) { - eprosima::fastrtps::Domain::removeSubscriber(subscriber_); + if (participant_) { + participant_->delete_subscriber(subscriber_); + } subscriber_ = nullptr; } @@ -137,7 +142,7 @@ void Manager::RemoveChangeListener(const ChangeConnection& conn) { local_conn.Disconnect(); } -bool Manager::CreatePublisher(RtpsParticipant* participant) { +bool Manager::CreatePublisher(eprosima::fastdds::dds::DomainParticipant* participant) { RtpsPublisherAttr pub_attr; RETURN_VAL_IF( !AttributesFiller::FillInPubAttr( @@ -148,7 +153,7 @@ bool Manager::CreatePublisher(RtpsParticipant* participant) { return publisher_ != nullptr; } -bool Manager::CreateSubscriber(RtpsParticipant* participant) { +bool Manager::CreateSubscriber(eprosima::fastdds::dds::DomainParticipant* participant) { RtpsSubscriberAttr sub_attr; RETURN_VAL_IF( !AttributesFiller::FillInSubAttr( diff --git a/cyber/service_discovery/specific_manager/manager.h b/cyber/service_discovery/specific_manager/manager.h index 76287be..b556002 100644 --- a/cyber/service_discovery/specific_manager/manager.h +++ b/cyber/service_discovery/specific_manager/manager.h @@ -24,6 +24,7 @@ #include "fastdds/dds/publisher/Publisher.hpp" #include "fastdds/dds/subscriber/Subscriber.hpp" +#include "fastdds/dds/domain/DomainParticipant.hpp" #include "xmlparser/attributes/PublisherAttributes.hpp" #include "xmlparser/attributes/SubscriberAttributes.hpp" @@ -53,7 +54,6 @@ class Manager { using ChangeFunc = std::function; using ChangeConnection = base::Connection; - using RtpsParticipant = eprosima::fastdds::rtps::RTPSParticipant; using RtpsPublisherAttr = eprosima::fastdds::PublisherAttributes; using RtpsSubscriberAttr = eprosima::fastdds::SubscriberAttributes; @@ -74,7 +74,7 @@ class Manager { * @return true if start successfully * @return false if start fail */ - bool StartDiscovery(RtpsParticipant* participant); + bool StartDiscovery(eprosima::fastdds::dds::DomainParticipant* participant); /** * @brief Stop topology discovery @@ -135,8 +135,8 @@ class Manager { int process_id) = 0; protected: - bool CreatePublisher(RtpsParticipant* participant); - bool CreateSubscriber(RtpsParticipant* participant); + bool CreatePublisher(eprosima::fastdds::dds::DomainParticipant* participant); + bool CreateSubscriber(eprosima::fastdds::dds::DomainParticipant* participant); virtual bool Check(const RoleAttributes& attr) = 0; virtual void Dispose(const ChangeMsg& msg) = 0; @@ -157,6 +157,7 @@ class Manager { std::string host_name_; int process_id_; std::string channel_name_; + eprosima::fastdds::dds::DomainParticipant* participant_; eprosima::fastdds::dds::Publisher* publisher_; std::mutex lock_; eprosima::fastdds::dds::Subscriber* subscriber_; diff --git a/cyber/transport/rtps/attributes_filler.cc b/cyber/transport/rtps/attributes_filler.cc index 3d480a7..40cf54c 100644 --- a/cyber/transport/rtps/attributes_filler.cc +++ b/cyber/transport/rtps/attributes_filler.cc @@ -35,7 +35,7 @@ bool AttributesFiller::FillInPubAttr( pub_attr->topic.topicName = channel_name; pub_attr->topic.topicDataType = "UnderlayMessage"; - pub_attr->topic.topicKind = eprosima::fastdds::dds::NO_KEY; + pub_attr->topic.topicKind = eprosima::fastdds::rtps::NO_KEY; switch (qos.history()) { case QosHistoryPolicy::HISTORY_KEEP_LAST: @@ -87,8 +87,8 @@ bool AttributesFiller::FillInPubAttr( // transform messages per second to rtps heartbeat // set default heartbeat period - pub_attr->times.heartbeatPeriod.seconds = 1; - pub_attr->times.heartbeatPeriod.fraction = 0; + pub_attr->times.heartbeat_period.seconds = 1; + pub_attr->times.heartbeat_period.fraction(0); if (qos.mps() != 0) { uint64_t mps = qos.mps(); @@ -103,14 +103,14 @@ bool AttributesFiller::FillInPubAttr( uint32_t fraction = fractions & 0xffffffff; int32_t seconds = static_cast(fractions >> 32); - pub_attr->times.heartbeatPeriod.seconds = seconds; - pub_attr->times.heartbeatPeriod.fraction = fraction; + pub_attr->times.heartbeat_period.seconds = seconds; + pub_attr->times.heartbeat_period.fraction(fraction); } pub_attr->qos.m_publishMode.kind = eprosima::fastdds::dds::ASYNCHRONOUS_PUBLISH_MODE; pub_attr->historyMemoryPolicy = - eprosima::fastdds::dds::DYNAMIC_RESERVE_MEMORY_MODE; + eprosima::fastdds::rtps::DYNAMIC_RESERVE_MEMORY_MODE; pub_attr->topic.resourceLimitsQos.max_samples = 10000; return true; @@ -122,7 +122,7 @@ bool AttributesFiller::FillInSubAttr( RETURN_VAL_IF_NULL(sub_attr, false); sub_attr->topic.topicName = channel_name; sub_attr->topic.topicDataType = "UnderlayMessage"; - sub_attr->topic.topicKind = eprosima::fastdds::dds::NO_KEY; + sub_attr->topic.topicKind = eprosima::fastdds::rtps::NO_KEY; switch (qos.history()) { case QosHistoryPolicy::HISTORY_KEEP_LAST: @@ -173,7 +173,7 @@ bool AttributesFiller::FillInSubAttr( } sub_attr->historyMemoryPolicy = - eprosima::fastdds::dds::DYNAMIC_RESERVE_MEMORY_MODE; + eprosima::fastdds::rtps::DYNAMIC_RESERVE_MEMORY_MODE; sub_attr->topic.resourceLimitsQos.max_samples = 10000; return true; diff --git a/cyber/transport/rtps/participant.cc b/cyber/transport/rtps/participant.cc index e619eb6..88b4e9d 100644 --- a/cyber/transport/rtps/participant.cc +++ b/cyber/transport/rtps/participant.cc @@ -16,7 +16,9 @@ #include "cyber/transport/rtps/participant.h" +#include "fastdds/dds/domain/DomainParticipantFactory.hpp" #include "fastdds/rtps/common/Locator.hpp" +#include "fastdds/utils/QosConverters.hpp" #include "xmlparser/attributes/ParticipantAttributes.hpp" #include "cyber/proto/transport_conf.pb.h" @@ -28,14 +30,17 @@ namespace apollo { namespace cyber { namespace transport { +using eprosima::fastdds::dds::DomainParticipantFactory; + Participant::Participant( const std::string& name, int send_port, - eprosima::fastdds::rtps::RTPSParticipantListener* listener) + eprosima::fastdds::dds::DomainParticipantListener* listener) : shutdown_(false), name_(name), send_port_(send_port), listener_(listener), - fastrtps_participant_(nullptr) {} + fastrtps_participant_(nullptr), + type_(new UnderlayMessageType()) {} Participant::~Participant() {} @@ -46,13 +51,13 @@ void Participant::Shutdown() { std::lock_guard lk(mutex_); if (fastrtps_participant_ != nullptr) { - eprosima::fastrtps::Domain::removeParticipant(fastrtps_participant_); + DomainParticipantFactory::get_instance()->delete_participant(fastrtps_participant_); fastrtps_participant_ = nullptr; listener_ = nullptr; } } -eprosima::fastdds::rtps::RTPSParticipant* Participant::fastrtps_participant() { +eprosima::fastdds::dds::DomainParticipant* Participant::fastrtps_participant() { if (shutdown_.load()) { return nullptr; } @@ -68,7 +73,7 @@ eprosima::fastdds::rtps::RTPSParticipant* Participant::fastrtps_participant() { void Participant::CreateFastRtpsParticipant( const std::string& name, int send_port, - eprosima::fastdds::rtps::RTPSParticipantListener* listener) { + eprosima::fastdds::dds::DomainParticipantListener* listener) { uint32_t domain_id = 80; const char* val = ::getenv("CYBER_DOMAIN_ID"); @@ -89,18 +94,16 @@ void Participant::CreateFastRtpsParticipant( } eprosima::fastdds::ParticipantAttributes attr; - attr.rtps.defaultSendPort = send_port; attr.rtps.port.domainIDGain = static_cast(part_attr_conf->domain_id_gain()); attr.rtps.port.portBase = static_cast(part_attr_conf->port_base()); - attr.rtps.use_IP6_to_send = false; - attr.rtps.builtin.use_SIMPLE_RTPSParticipantDiscoveryProtocol = true; - attr.rtps.builtin.use_SIMPLE_EndpointDiscoveryProtocol = true; - attr.rtps.builtin.m_simpleEDP.use_PublicationReaderANDSubscriptionWriter = + attr.rtps.builtin.discovery_config.discoveryProtocol = eprosima::fastdds::rtps::DiscoveryProtocol::SIMPLE; + attr.rtps.builtin.discovery_config.use_SIMPLE_EndpointDiscoveryProtocol = true; + attr.rtps.builtin.discovery_config.m_simpleEDP.use_PublicationReaderANDSubscriptionWriter = true; - attr.rtps.builtin.m_simpleEDP.use_PublicationWriterANDSubscriptionReader = + attr.rtps.builtin.discovery_config.m_simpleEDP.use_PublicationWriterANDSubscriptionReader = true; - attr.rtps.builtin.domainId = domain_id; + attr.domainId = domain_id; /** * The user should set the lease_duration and the announcement_period with @@ -108,8 +111,8 @@ void Participant::CreateFastRtpsParticipant( * cause the failure of the writer liveliness assertion in networks with high * latency or with lots of communication errors. */ - attr.rtps.builtin.leaseDuration.seconds = part_attr_conf->lease_duration(); - attr.rtps.builtin.leaseDuration_announcementperiod.seconds = + attr.rtps.builtin.discovery_config.leaseDuration.seconds = part_attr_conf->lease_duration(); + attr.rtps.builtin.discovery_config.leaseDuration_announcementperiod.seconds = part_attr_conf->announcement_period(); attr.rtps.setName(name.c_str()); @@ -126,22 +129,25 @@ void Participant::CreateFastRtpsParticipant( ADEBUG << "cyber ip: " << ip_env; eprosima::fastdds::rtps::Locator_t locator; - locator.port = 0; - RETURN_IF(!locator.set_IP4_address(ip_env)); + locator.port = send_port; + RETURN_IF(!eprosima::fastdds::rtps::IPLocator::setIPv4(locator, ip_env)); locator.kind = LOCATOR_KIND_UDPv4; attr.rtps.defaultUnicastLocatorList.push_back(locator); - attr.rtps.defaultOutLocatorList.push_back(locator); attr.rtps.builtin.metatrafficUnicastLocatorList.push_back(locator); - locator.set_IP4_address(239, 255, 0, 1); + eprosima::fastdds::rtps::IPLocator::setIPv4(locator, std::string("239.255.0.1")); attr.rtps.builtin.metatrafficMulticastLocatorList.push_back(locator); + eprosima::fastdds::dds::DomainParticipantExtendedQos extended_qos; + eprosima::fastdds::dds::utils::set_attributes_from_extended_qos( + attr, extended_qos + ); fastrtps_participant_ = - eprosima::fastrtps::Domain::createParticipant(attr, listener); + DomainParticipantFactory::get_instance()->create_participant(extended_qos, listener_); RETURN_IF_NULL(fastrtps_participant_); - eprosima::fastrtps::Domain::registerType(fastrtps_participant_, &type_); + type_.register_type(fastrtps_participant_); } } // namespace transport diff --git a/cyber/transport/rtps/participant.h b/cyber/transport/rtps/participant.h index 40388bb..cd9687b 100644 --- a/cyber/transport/rtps/participant.h +++ b/cyber/transport/rtps/participant.h @@ -22,7 +22,11 @@ #include #include +#include "fastdds/dds/domain/DomainParticipant.hpp" +#include "fastdds/dds/topic/TypeSupport.hpp" #include "fastdds/rtps/common/Locator.hpp" +#include "fastdds/rtps/participant/RTPSParticipant.hpp" +#include "fastdds/rtps/participant/RTPSParticipantListener.hpp" #include "cyber/transport/rtps/underlay_message_type.h" @@ -36,25 +40,25 @@ using ParticipantPtr = std::shared_ptr; class Participant { public: Participant(const std::string& name, int send_port, - eprosima::fastdds::rtps::RTPSParticipantListener* listener = nullptr); + eprosima::fastdds::dds::DomainParticipantListener* listener = nullptr); virtual ~Participant(); void Shutdown(); - eprosima::fastdds::rtps::RTPSParticipant* fastrtps_participant(); + eprosima::fastdds::dds::DomainParticipant* fastrtps_participant(); bool is_shutdown() const { return shutdown_.load(); } private: void CreateFastRtpsParticipant( const std::string& name, int send_port, - eprosima::fastdds::rtps::RTPSParticipantListener* listener); + eprosima::fastdds::dds::DomainParticipantListener* listener); std::atomic shutdown_; std::string name_; int send_port_; - eprosima::fastdds::rtps::RTPSParticipantListener* listener_; - UnderlayMessageType type_; - eprosima::fastdds::rtps::RTPSParticipant* fastrtps_participant_; + eprosima::fastdds::dds::DomainParticipantListener* listener_; + eprosima::fastdds::dds::TypeSupport type_; + eprosima::fastdds::dds::DomainParticipant* fastrtps_participant_; std::mutex mutex_; }; diff --git a/cyber/transport/rtps/sub_listener.cc b/cyber/transport/rtps/sub_listener.cc index 9fd907a..7757a62 100644 --- a/cyber/transport/rtps/sub_listener.cc +++ b/cyber/transport/rtps/sub_listener.cc @@ -28,18 +28,19 @@ SubListener::SubListener(const NewMsgCallback& callback) SubListener::~SubListener() {} -void SubListener::onNewDataMessage(eprosima::fastdds::dds::Subscribe* sub) { - RETURN_IF_NULL(sub); +void SubListener::on_data_available(eprosima::fastdds::dds::DataReader* reader) { + RETURN_IF_NULL(reader); RETURN_IF_NULL(callback_); std::lock_guard lock(mutex_); // fetch channel name - auto channel_id = common::Hash(sub->getAttributes().topic.getTopicName()); - eprosima::fastrtps::SampleInfo_t m_info; + auto channel_id = common::Hash(reader->get_topicdescription()->get_name()); + eprosima::fastdds::dds::SampleInfo m_info; UnderlayMessage m; - RETURN_IF(!sub->takeNextData(reinterpret_cast(&m), &m_info)); - RETURN_IF(m_info.sampleKind != eprosima::fastdds::rtps::ALIVE); + RETURN_IF( + reader->take_next_sample(reinterpret_cast(&m), &m_info) != eprosima::fastdds::dds::RETCODE_OK); + RETURN_IF(m_info.instance_state != eprosima::fastdds::dds::ALIVE_INSTANCE_STATE); // fetch MessageInfo char* ptr = @@ -65,13 +66,6 @@ void SubListener::onNewDataMessage(eprosima::fastdds::dds::Subscribe* sub) { callback_(channel_id, msg_str, msg_info_); } -void SubListener::onSubscriptionMatched( - eprosima::fastdds::dds::Subscribe* sub, - eprosima::fastdds::rtps::MatchingInfo& info) { - (void)sub; - (void)info; -} - } // namespace transport } // namespace cyber } // namespace apollo diff --git a/cyber/transport/rtps/sub_listener.h b/cyber/transport/rtps/sub_listener.h index 028b708..0e5d425 100644 --- a/cyber/transport/rtps/sub_listener.h +++ b/cyber/transport/rtps/sub_listener.h @@ -23,8 +23,12 @@ #include #include -#include "fastdds/dds/subscriber/Subscriber.hpp" -#include "fastdds/dds/subscriber/SubscriberListener.hpp" +#include "fastdds/dds/subscriber/SampleInfo.hpp" +#include "fastdds/dds/subscriber/DataReader.hpp" +#include "fastdds/dds/subscriber/DataReaderListener.hpp" +#include "fastdds/dds/topic/TopicDescription.hpp" +// #include "fastdds/dds/subscriber/Subscriber.hpp" +// #include "fastdds/dds/subscriber/SubscriberListener.hpp" #include "fastdds/rtps/common/MatchingInfo.hpp" #include "cyber/transport/message/message_info.h" @@ -38,7 +42,7 @@ namespace transport { class SubListener; using SubListenerPtr = std::shared_ptr; -class SubListener : public eprosima::fastdds::dds::SubscriberListener { +class SubListener : public eprosima::fastdds::dds::DataReaderListener { public: using NewMsgCallback = std::function& msg_str, @@ -47,10 +51,7 @@ class SubListener : public eprosima::fastdds::dds::SubscriberListener { explicit SubListener(const NewMsgCallback& callback); virtual ~SubListener(); - void onNewDataMessage(eprosima::fastdds::dds::Subscriber* sub); - void onSubscriptionMatched( - eprosima::fastdds::dds::Subscribe* sub, - eprosima::fastdds::rtps::MatchingInfo& info); // NOLINT + virtual void on_data_available(eprosima::fastdds::dds::DataReader* reader) override; private: NewMsgCallback callback_; diff --git a/cyber/transport/rtps/underlay_message_type.cc b/cyber/transport/rtps/underlay_message_type.cc index 8af2601..989e4bf 100644 --- a/cyber/transport/rtps/underlay_message_type.cc +++ b/cyber/transport/rtps/underlay_message_type.cc @@ -47,14 +47,15 @@ UnderlayMessageType::~UnderlayMessageType() { } } -bool UnderlayMessageType::serialize(void* data, SerializedPayload_t* payload) { - UnderlayMessage* p_type = reinterpret_cast(data); +bool UnderlayMessageType::serialize( + const void* const data, eprosima::fastdds::rtps::SerializedPayload_t* payload) { + const UnderlayMessage* p_type = reinterpret_cast(data); eprosima::fastcdr::FastBuffer fastbuffer( reinterpret_cast(payload->data), payload->max_size); // Object that manages the raw buffer. eprosima::fastcdr::Cdr ser( fastbuffer, eprosima::fastcdr::Cdr::DEFAULT_ENDIAN, - eprosima::fastcdr::Cdr::DDS_CDR); // Object that serializes the data. + eprosima::fastcdr::CdrVersion::DDS_CDR); // Object that serializes the data. payload->encapsulation = ser.endianness() == eprosima::fastcdr::Cdr::BIG_ENDIANNESS ? CDR_BE : CDR_LE; @@ -62,12 +63,12 @@ bool UnderlayMessageType::serialize(void* data, SerializedPayload_t* payload) { ser.serialize_encapsulation(); p_type->serialize(ser); // Serialize the object: payload->length = - (uint32_t)ser.getSerializedDataLength(); // Get the serialized length + (uint32_t)ser.get_serialized_data_length(); // Get the serialized length return true; } -bool UnderlayMessageType::deserialize(SerializedPayload_t* payload, - void* data) { +bool UnderlayMessageType::deserialize( + eprosima::fastdds::rtps::SerializedPayload_t* payload, void* data) { UnderlayMessage* p_type = reinterpret_cast( data); // Convert DATA to pointer of your type eprosima::fastcdr::FastBuffer fastbuffer( @@ -75,7 +76,7 @@ bool UnderlayMessageType::deserialize(SerializedPayload_t* payload, payload->length); // Object that manages the raw buffer. eprosima::fastcdr::Cdr deser( fastbuffer, eprosima::fastcdr::Cdr::DEFAULT_ENDIAN, - eprosima::fastcdr::Cdr::DDS_CDR); // Object that deserializes the data. + eprosima::fastcdr::CdrVersion::DDS_CDR); // Object that deserializes the data. // Deserialize encapsulation. deser.read_encapsulation(); payload->encapsulation = @@ -86,10 +87,10 @@ bool UnderlayMessageType::deserialize(SerializedPayload_t* payload, } std::function UnderlayMessageType::getSerializedSizeProvider( - void* data) { + const void* const data) { return [data]() -> uint32_t { return (uint32_t)type::getCdrSerializedSize( - *static_cast(data)) + + *static_cast(data)) + 4 /*encapsulation*/; }; } @@ -102,9 +103,12 @@ void UnderlayMessageType::deleteData(void* data) { delete (reinterpret_cast(data)); } -bool UnderlayMessageType::getKey(void* data, InstanceHandle_t* handle) { +bool UnderlayMessageType::getKey( + const void* const data, + eprosima::fastdds::rtps::InstanceHandle_t* handle, + bool force_md5) { RETURN_VAL_IF((!m_isGetKeyDefined), false); - UnderlayMessage* p_type = reinterpret_cast(data); + const UnderlayMessage* p_type = reinterpret_cast(data); eprosima::fastcdr::FastBuffer fastbuffer( reinterpret_cast(m_keyBuffer), UnderlayMessage::getKeyMaxCdrSerializedSize()); // Object that manages @@ -116,7 +120,7 @@ bool UnderlayMessageType::getKey(void* data, InstanceHandle_t* handle) { p_type->serializeKey(ser); if (UnderlayMessage::getKeyMaxCdrSerializedSize() > 16) { m_md5.init(); - m_md5.update(m_keyBuffer, (unsigned int)ser.getSerializedDataLength()); + m_md5.update(m_keyBuffer, (unsigned int)ser.get_serialized_data_length()); m_md5.finalize(); for (uint8_t i = 0; i < 16; ++i) { handle->value[i] = m_md5.digest[i]; diff --git a/cyber/transport/rtps/underlay_message_type.h b/cyber/transport/rtps/underlay_message_type.h index 194b3c2..20585db 100644 --- a/cyber/transport/rtps/underlay_message_type.h +++ b/cyber/transport/rtps/underlay_message_type.h @@ -35,13 +35,16 @@ class UnderlayMessageType : public eprosima::fastdds::dds::TopicDataType { UnderlayMessageType(); virtual ~UnderlayMessageType(); - bool serialize(void* data, SerializedPayload_t* payload); - bool deserialize(SerializedPayload_t* payload, void* data); - std::function getSerializedSizeProvider(void* data); - bool getKey(void* data, InstanceHandle_t* ihandle); + virtual bool serialize(const void* const data, eprosima::fastdds::rtps::SerializedPayload_t* payload) override; + virtual bool deserialize(eprosima::fastdds::rtps::SerializedPayload_t* payload, void* data) override; + virtual std::function getSerializedSizeProvider(const void* const data) override; + virtual bool getKey( + const void* const data, + eprosima::fastdds::rtps::InstanceHandle_t* ihandle, + bool force_md5 = false) override; void* createData(); void deleteData(void* data); - MD5 m_md5; + eprosima::fastdds::MD5 m_md5; unsigned char* m_keyBuffer; };