proto定义
syntax = "proto3";package example;message MyMessage {int32 id = 1;string content = 2;
}
pub端
#include <iostream>
#include <fastdds/dds/domain/DomainParticipantFactory.hpp>
#include <fastdds/dds/domain/DomainParticipant.hpp>
#include <fastdds/dds/topic/Topic.hpp>
#include <fastdds/dds/publisher/Publisher.hpp>
#include <fastdds/dds/publisher/DataWriter.hpp>
#include <fastdds/dds/publisher/qos/PublisherQos.hpp>
#include <fastdds/dds/publisher/qos/DataWriterQos.hpp>
#include <fastdds/dds/topic/TypeSupport.hpp>
#include "example.pb.h"using namespace eprosima::fastdds::dds;class ProtobufType : public TopicDataType {
public:ProtobufType() {setName("example::MyMessage");m_typeSize = 4 + 100; // Adjust size accordinglym_isGetKeyDefined = false;}bool serialize(void* data, eprosima::fastrtps::rtps::SerializedPayload_t* payload) override {example::MyMessage* message = static_cast<example::MyMessage*>(data);std::string serializedMessage;message->SerializeToString(&serializedMessage);payload->length = static_cast<uint32_t>(serializedMessage.size());memcpy(payload->data, serializedMessage.data(), payload->length);return true;}bool deserialize(eprosima::fastrtps::rtps::SerializedPayload_t* payload, void* data) override {example::MyMessage* message = static_cast<example::MyMessage*>(data);std::string serializedMessage(reinterpret_cast<char*>(payload->data), payload->length);return message->ParseFromString(serializedMessage);}std::function<uint32_t()> getSerializedSizeProvider(void* data) override {return [data]() -> uint32_t {example::MyMessage* message = static_cast<example::MyMessage*>(data);return static_cast<uint32_t>(message->ByteSizeLong());};}void* createData() override {return new example::MyMessage();}void deleteData(void* data) override {delete static_cast<example::MyMessage*>(data);}bool getKey(void* data, InstanceHandle_t* ihandle, bool force_md5 = false) override {return false;}
};int main() {// 创建DomainParticipantDomainParticipantQos participantQos;DomainParticipant* participant = DomainParticipantFactory::get_instance()->create_participant(0, participantQos);if (participant == nullptr) {std::cerr << "创建DomainParticipant失败!" << std::endl;return 1;}// 注册类型TypeSupport myType(new ProtobufType());myType.register_type(participant);// 创建TopicTopicQos topicQos;Topic* topic = participant->create_topic("MyTopic", "example::MyMessage", topicQos);if (topic == nullptr) {std::cerr << "创建Topic失败!" << std::endl;return 1;}// 创建PublisherPublisherQos publisherQos;Publisher* publisher = participant->create_publisher(publisherQos);if (publisher == nullptr) {std::cerr << "创建Publisher失败!" << std::endl;return 1;}// 创建DataWriterDataWriterQos writerQos;DataWriter* writer = publisher->create_datawriter(topic, writerQos);if (writer == nullptr) {std::cerr << "创建DataWriter失败!" << std::endl;return 1;}std::this_thread::sleep_for(std::chrono::seconds(1));// 创建Protobuf消息example::MyMessage message;message.set_id(1);message.set_content("123");// 发布消息writer->write(&message);// 清理资源participant->delete_contained_entities();DomainParticipantFactory::get_instance()->delete_participant(participant);return 0;
}
SUB端
#include <iostream>
#include <thread>
#include <chrono>
#include <fastdds/dds/domain/DomainParticipantFactory.hpp>
#include <fastdds/dds/domain/DomainParticipant.hpp>
#include <fastdds/dds/topic/Topic.hpp>
#include <fastdds/dds/subscriber/Subscriber.hpp>
#include <fastdds/dds/subscriber/DataReader.hpp>
#include <fastdds/dds/subscriber/qos/SubscriberQos.hpp>
#include <fastdds/dds/subscriber/qos/DataReaderQos.hpp>
#include <fastdds/dds/subscriber/SampleInfo.hpp>
#include <fastdds/dds/topic/TypeSupport.hpp>
#include "example.pb.h"using namespace eprosima::fastdds::dds;class MyListener : public DataReaderListener {void on_data_available(DataReader* reader) override {example::MyMessage serializedMessage;SampleInfo info;// 读取数据if (reader->take_next_sample(&serializedMessage, &info) == ReturnCode_t::RETCODE_OK) {if (info.valid_data) {// 打印消息std::cout << "Received message: id=" << serializedMessage.id() << ", content=" << serializedMessage.content() << std::endl;}}}
};class ProtobufType : public TopicDataType {
public:ProtobufType() {setName("example::MyMessage");m_typeSize = 4 + 100; // Adjust size accordinglym_isGetKeyDefined = false;}bool serialize(void* data, eprosima::fastrtps::rtps::SerializedPayload_t* payload) override {example::MyMessage* message = static_cast<example::MyMessage*>(data);std::string serializedMessage;message->SerializeToString(&serializedMessage);payload->length = static_cast<uint32_t>(serializedMessage.size());memcpy(payload->data, serializedMessage.data(), payload->length);return true;}bool deserialize(eprosima::fastrtps::rtps::SerializedPayload_t* payload, void* data) override {example::MyMessage* message = static_cast<example::MyMessage*>(data);std::string serializedMessage(reinterpret_cast<char*>(payload->data), payload->length-1);return message->ParseFromString(serializedMessage);}std::function<uint32_t()> getSerializedSizeProvider(void* data) override {return [data]() -> uint32_t {example::MyMessage* message = static_cast<example::MyMessage*>(data);return static_cast<uint32_t>(message->ByteSizeLong());};}void* createData() override {return new example::MyMessage();}void deleteData(void* data) override {delete static_cast<example::MyMessage*>(data);}bool getKey(void* data, eprosima::fastrtps::rtps::InstanceHandle_t* ihandle, bool force_md5 = false) override {// 对于此示例，我们假设不需要键（即，我们不使用基于键的过滤） //*ihandle = eprosima::fastrtps::rtps::c_InstanceHandle_Unknown();return false;}};int main() {// 创建DomainParticipantDomainParticipantQos participantQos;DomainParticipant* participant = DomainParticipantFactory::get_instance()->create_participant(0, participantQos);if (participant == nullptr) {std::cerr << "创建DomainParticipant失败!" << std::endl;return 1;}// 注册类型TypeSupport myType(new ProtobufType());myType.register_type(participant);// 创建TopicTopicQos topicQos;Topic* topic = participant->create_topic("MyTopic", "example::MyMessage", topicQos);if (topic == nullptr) {std::cerr << "创建Topic失败!" << std::endl;return 1;}// 创建SubscriberSubscriberQos subscriberQos;Subscriber* subscriber = participant->create_subscriber(subscriberQos);if (subscriber == nullptr) {std::cerr << "创建Subscriber失败!" << std::endl;return 1;}// 创建DataReaderDataReaderQos readerQos;DataReader* reader = subscriber->create_datareader(topic, readerQos, new MyListener());if (reader == nullptr) {std::cerr << "创建DataReader失败!" << std::endl;return 1;}// 等待接收消息std::cout << "Waiting for messages..." << std::endl;std::this_thread::sleep_for(std::chrono::minutes(100));// 清理资源participant->delete_contained_entities();DomainParticipantFactory::get_instance()->delete_participant(participant);return 0;
}
