From 0b54ae17a1adbeb2799730523e3425639d995cb7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=82=B5=E6=9C=B1=E9=93=AD?= Date: Sun, 20 Sep 2026 09:31:32 +0800 Subject: [PATCH] =?UTF-8?q?=E4=B8=8A=E4=BC=A0=E6=96=87=E4=BB=B6=E8=87=B3?= =?UTF-8?q?=20'WeighUI/dds'?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: 邵朱铭 --- WeighUI/dds/Subscriber.cxx | 175 +++++++++++++++++++++++++++++++++++++ WeighUI/dds/Subscriber.hpp | 52 +++++++++++ WeighUI/dds/msg.hpp | 16 ++++ 3 files changed, 243 insertions(+) create mode 100644 WeighUI/dds/Subscriber.cxx create mode 100644 WeighUI/dds/Subscriber.hpp create mode 100644 WeighUI/dds/msg.hpp diff --git a/WeighUI/dds/Subscriber.cxx b/WeighUI/dds/Subscriber.cxx new file mode 100644 index 0000000..d361d01 --- /dev/null +++ b/WeighUI/dds/Subscriber.cxx @@ -0,0 +1,175 @@ +#include "Subscriber.hpp" + +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +#include +#include "msg.hpp" + +using namespace eprosima::fastdds::dds; + +DdsSubscriberApp::DdsSubscriberApp(const int& domain_id) + : factory_(nullptr) + , participant_(nullptr) + , subscriber_(nullptr) + , summary_topic_(nullptr) + , summary_reader_(nullptr) + , summary_type_(new WeighingSystem::SummaryUpdatePubSubType()) + , queuesuccess_topic_(nullptr) + , queuesuccess_reader_(nullptr) + , queuesuccess_type_(new Queue::QueueSuccessPubSubType()) + , stop_(false) +{ + DomainParticipantQos pqos = PARTICIPANT_QOS_DEFAULT; + pqos.name("WeighUI_sub_participant"); + pqos.wire_protocol().builtin.discovery_config.leaseDuration = Duration_t(60, 0); + pqos.wire_protocol().builtin.discovery_config.leaseDuration_announcementperiod = Duration_t(30, 0); + factory_ = DomainParticipantFactory::get_shared_instance(); + participant_ = factory_->create_participant(domain_id, pqos, nullptr, StatusMask::none()); + if (participant_ == nullptr) + { + throw std::runtime_error("WeighUI Participant initialization failed"); + } + + SubscriberQos sub_qos = SUBSCRIBER_QOS_DEFAULT; + participant_->get_default_subscriber_qos(sub_qos); + subscriber_ = participant_->create_subscriber(sub_qos, nullptr, StatusMask::none()); + if (subscriber_ == nullptr) + { + throw std::runtime_error("WeighUI Subscriber initialization failed"); + } + + TopicQos topic_qos = TOPIC_QOS_DEFAULT; + participant_->get_default_topic_qos(topic_qos); + + DataReaderQos reader_qos = DATAREADER_QOS_DEFAULT; + subscriber_->get_default_datareader_qos(reader_qos); + reader_qos.reliability().kind = ReliabilityQosPolicyKind::RELIABLE_RELIABILITY_QOS; + reader_qos.reliability().max_blocking_time = Duration_t(1, 0); + reader_qos.durability().kind = DurabilityQosPolicyKind::VOLATILE_DURABILITY_QOS; + reader_qos.history().kind = HistoryQosPolicyKind::KEEP_LAST_HISTORY_QOS; + reader_qos.history().depth = 1; + reader_qos.resource_limits().max_samples = 200; + reader_qos.resource_limits().max_instances = 1; + reader_qos.resource_limits().max_samples_per_instance = 100; + reader_qos.data_sharing().off(); + + // Register SummaryUpdate type and create reader + summary_type_.register_type(participant_); + qDebug() << "[WeighUI] registered SummaryUpdate type:" + << summary_type_.get_type_name().c_str(); + + summary_topic_ = participant_->create_topic("SummaryUpdate", summary_type_.get_type_name(), topic_qos); + if (summary_topic_ == nullptr) + { + throw std::runtime_error("WeighUI::SummaryUpdate Topic initialization failed"); + } + + summary_reader_ = subscriber_->create_datareader(summary_topic_, reader_qos, this, StatusMask::all()); + if (summary_reader_ == nullptr) + { + throw std::runtime_error("WeighUI::SummaryUpdate DataReader initialization failed"); + } + + // Register QueueSuccess topic and reader (排队成功 → display_text) + queuesuccess_type_.register_type(participant_); + qDebug() << "[WeighUI] registered QueueSuccess type:" + << queuesuccess_type_.get_type_name().c_str(); + + queuesuccess_topic_ = participant_->create_topic("QueueSuccess", queuesuccess_type_.get_type_name(), topic_qos); + if (queuesuccess_topic_ == nullptr) + { + throw std::runtime_error("WeighUI::QueueSuccess Topic initialization failed"); + } + + queuesuccess_reader_ = subscriber_->create_datareader(queuesuccess_topic_, reader_qos, this, StatusMask::all()); + if (queuesuccess_reader_ == nullptr) + { + throw std::runtime_error("WeighUI::QueueSuccess DataReader initialization failed"); + } +} + +DdsSubscriberApp::~DdsSubscriberApp() +{ + if (nullptr != participant_) + { + participant_->delete_contained_entities(); + factory_->delete_participant(participant_); + } +} + +void DdsSubscriberApp::on_subscription_matched( + DataReader* reader, + const SubscriptionMatchedStatus& info) +{ + std::string topic = reader->get_topicdescription()->get_name(); + qDebug() << "[WeighUI] SubscriptionMatched topic=" << topic.c_str() + << " current_count=" << info.current_count + << " change=" << info.current_count_change + << " total_count=" << info.total_count; +} + +void DdsSubscriberApp::on_data_available(DataReader* reader) +{ + SampleInfo info; + std::string topic_name = reader->get_topicdescription()->get_name(); + + qDebug() << "[WeighUI] on_data_available topic=" << topic_name.c_str(); + + if (topic_name == "SummaryUpdate") + { + WeighingSystem::SummaryUpdate sample_; + while ((!is_stopped()) && (RETCODE_OK == reader->take_next_sample(&sample_, &info))) + { + if ((info.instance_state == ALIVE_INSTANCE_STATE) && info.valid_data) + { + std::unique_lock lock(DdsMsgData::queue_cv_mtx_); + DdsMsgData::SummaryUpdate_queue_.push(std::move(sample_)); + lock.unlock(); + } + } + } + else if (topic_name == "QueueSuccess") + { + Queue::QueueSuccess sample_; + while ((!is_stopped()) && (RETCODE_OK == reader->take_next_sample(&sample_, &info))) + { + if ((info.instance_state == ALIVE_INSTANCE_STATE) && info.valid_data) + { + std::unique_lock lock(DdsMsgData::queue_cv_mtx_); + DdsMsgData::QueueSuccess_queue_.push(std::move(sample_)); + lock.unlock(); + } + } + } +} + +void DdsSubscriberApp::run() +{ + while (!is_stopped()) + { + std::unique_lock period_lock(terminate_cv_mtx_); + terminate_cv_.wait_for(period_lock, std::chrono::milliseconds(period_ms_), + [this]() { return is_stopped(); }); + } +} + +bool DdsSubscriberApp::is_stopped() +{ + return stop_.load(); +} + +void DdsSubscriberApp::stop() +{ + stop_.store(true); + terminate_cv_.notify_all(); +} \ No newline at end of file diff --git a/WeighUI/dds/Subscriber.hpp b/WeighUI/dds/Subscriber.hpp new file mode 100644 index 0000000..964e180 --- /dev/null +++ b/WeighUI/dds/Subscriber.hpp @@ -0,0 +1,52 @@ +#ifndef WEIGHUI_DDS_SUBSCRIBER_HPP +#define WEIGHUI_DDS_SUBSCRIBER_HPP + +#include +#include +#include +#include + +#include +#include +#include +#include + +#include "WeighingDDSTypePubSubTypes.hpp" + +class DdsSubscriberApp : public eprosima::fastdds::dds::DataReaderListener +{ +public: + DdsSubscriberApp(const int& domain_id); + virtual ~DdsSubscriberApp(); + + void on_data_available(eprosima::fastdds::dds::DataReader* reader) override; + void on_subscription_matched( + eprosima::fastdds::dds::DataReader* reader, + const eprosima::fastdds::dds::SubscriptionMatchedStatus& info) override; + + void run(); + void stop(); + +private: + bool is_stopped(); + + std::shared_ptr factory_; + eprosima::fastdds::dds::DomainParticipant* participant_; + eprosima::fastdds::dds::Subscriber* subscriber_; + + eprosima::fastdds::dds::Topic* summary_topic_; + eprosima::fastdds::dds::DataReader* summary_reader_; + eprosima::fastdds::dds::TypeSupport summary_type_; + + // 排队成功 (上位机发布) → WeighUI 取 display_text 显示 + eprosima::fastdds::dds::Topic* queuesuccess_topic_; + eprosima::fastdds::dds::DataReader* queuesuccess_reader_; + eprosima::fastdds::dds::TypeSupport queuesuccess_type_; + + std::atomic stop_; + uint32_t period_ms_ = 100; + mutable std::mutex terminate_cv_mtx_; + std::condition_variable terminate_cv_; +}; + +#endif \ No newline at end of file diff --git a/WeighUI/dds/msg.hpp b/WeighUI/dds/msg.hpp new file mode 100644 index 0000000..fdf302c --- /dev/null +++ b/WeighUI/dds/msg.hpp @@ -0,0 +1,16 @@ +#ifndef WEIGHUI_DDS_MSG_HPP +#define WEIGHUI_DDS_MSG_HPP + +#include +#include +#include "WeighingDDSType.hpp" + +class DdsMsgData { +public: + static std::queue SummaryUpdate_queue_; + // 排队成功 → 显示 display_text + static std::queue QueueSuccess_queue_; + static std::mutex queue_cv_mtx_; +}; + +#endif \ No newline at end of file