diff --git a/WeighUI/dds/Subscriber.cxx b/WeighUI/dds/Subscriber.cxx new file mode 100644 index 0000000..7965143 --- /dev/null +++ b/WeighUI/dds/Subscriber.cxx @@ -0,0 +1,238 @@ +#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) + , weightinfook_topic_(nullptr) + , weightinfook_reader_(nullptr) + , weightinfook_type_(new WeighingSystem::WeightInfoOkPubSubType()) + , weightinfoerror_topic_(nullptr) + , weightinfoerror_reader_(nullptr) + , weightinfoerror_type_(new WeighingSystem::WeightInfoErrorPubSubType()) + , summary_topic_(nullptr) + , summary_reader_(nullptr) + , summary_type_(new WeighingSystem::SummaryUpdatePubSubType()) + , scaleinfo_topic_(nullptr) + , scaleinfo_reader_(nullptr) + , scaleinfo_type_(new WeighingSystem::ScaleInfoPubSubType()) + , 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"); + } + + weightinfook_type_.register_type(participant_); + qDebug() << "[WeighUI] registered WeightInfoOk type:" + << weightinfook_type_.get_type_name().c_str(); + + 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); + weightinfook_topic_ = participant_->create_topic("WeightInfoOk", weightinfook_type_.get_type_name(), topic_qos); + if (weightinfook_topic_ == nullptr) + { + throw std::runtime_error("WeighUI::WeightInfoOk Topic initialization failed"); + } + + 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(); + weightinfook_reader_ = subscriber_->create_datareader(weightinfook_topic_, reader_qos, this, StatusMask::all()); + if (weightinfook_reader_ == nullptr) + { + throw std::runtime_error("WeighUI::WeightInfoOk DataReader initialization failed"); + } + + // Register WeightInfoError type and create reader (Warning 主题, 弹窗样式与 WeightInfoOk 完全一致) + weightinfoerror_type_.register_type(participant_); + qDebug() << "[WeighUI] registered WeightInfoError type:" + << weightinfoerror_type_.get_type_name().c_str(); + + weightinfoerror_topic_ = participant_->create_topic("WeightInfoError", weightinfoerror_type_.get_type_name(), topic_qos); + if (weightinfoerror_topic_ == nullptr) + { + throw std::runtime_error("WeighUI::WeightInfoError Topic initialization failed"); + } + + weightinfoerror_reader_ = subscriber_->create_datareader(weightinfoerror_topic_, reader_qos, this, StatusMask::all()); + if (weightinfoerror_reader_ == nullptr) + { + throw std::runtime_error("WeighUI::WeightInfoError DataReader initialization failed"); + } + + // 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 ScaleInfo type and create reader (实时重量) + scaleinfo_type_.register_type(participant_); + qDebug() << "[WeighUI] registered ScaleInfo type:" + << scaleinfo_type_.get_type_name().c_str(); + + scaleinfo_topic_ = participant_->create_topic("ScaleInfo", scaleinfo_type_.get_type_name(), topic_qos); + if (scaleinfo_topic_ == nullptr) + { + throw std::runtime_error("WeighUI::ScaleInfo Topic initialization failed"); + } + + scaleinfo_reader_ = subscriber_->create_datareader(scaleinfo_topic_, reader_qos, this, StatusMask::all()); + if (scaleinfo_reader_ == nullptr) + { + throw std::runtime_error("WeighUI::ScaleInfo 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 == "WeightInfoOk") + { + WeighingSystem::WeightInfoOk 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::WeightInfoOk_queue_.push(std::move(sample_)); + lock.unlock(); + } + } + } + else if (topic_name == "WeightInfoError") + { + WeighingSystem::WeightInfoError 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::WeightInfoError_queue_.push(std::move(sample_)); + lock.unlock(); + } + } + } + else 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 == "ScaleInfo") + { + WeighingSystem::ScaleInfo 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::ScaleInfo_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..403ebfb --- /dev/null +++ b/WeighUI/dds/Subscriber.hpp @@ -0,0 +1,61 @@ +#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* weightinfook_topic_; + eprosima::fastdds::dds::DataReader* weightinfook_reader_; + eprosima::fastdds::dds::TypeSupport weightinfook_type_; + + // WeightInfoError 主题订阅 (Warning 主题, 弹窗样式与 WeightInfoOk 完全一致) + eprosima::fastdds::dds::Topic* weightinfoerror_topic_; + eprosima::fastdds::dds::DataReader* weightinfoerror_reader_; + eprosima::fastdds::dds::TypeSupport weightinfoerror_type_; + + eprosima::fastdds::dds::Topic* summary_topic_; + eprosima::fastdds::dds::DataReader* summary_reader_; + eprosima::fastdds::dds::TypeSupport summary_type_; + + // 实时重量 (weigh 程序发布) + eprosima::fastdds::dds::Topic* scaleinfo_topic_; + eprosima::fastdds::dds::DataReader* scaleinfo_reader_; + eprosima::fastdds::dds::TypeSupport scaleinfo_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..42cb17c --- /dev/null +++ b/WeighUI/dds/msg.hpp @@ -0,0 +1,19 @@ +#ifndef WEIGHUI_DDS_MSG_HPP +#define WEIGHUI_DDS_MSG_HPP + +#include +#include +#include "WeighingDDSType.hpp" + +class DdsMsgData { +public: + static std::queue WeightInfoOk_queue_; + // 称重业务异常信息 (Warning 主题, 弹窗样式与 WeightInfoOk 完全一致) + static std::queue WeightInfoError_queue_; + static std::queue SummaryUpdate_queue_; + // 实时重量 (来自 weigh 程序的 ScaleInfo 主题) + static std::queue ScaleInfo_queue_; + static std::mutex queue_cv_mtx_; +}; + +#endif \ No newline at end of file