From 47336c9a07703786f0a63b628cb8321a1bc23ac2 Mon Sep 17 00:00:00 2001 From: baocm Date: Mon, 12 Jan 2026 21:21:51 +0800 Subject: [PATCH] =?UTF-8?q?1.=20=E5=A2=9E=E5=8A=A0=E9=80=BB=E8=BE=91?= =?UTF-8?q?=E5=A4=84=E7=90=86=E7=A8=8B=E5=BA=8F=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CMakeLists.txt | 13 + core/Publisher.cxx | 518 ++++++++++++++++++++++++++++++++ core/Publisher.hpp | 85 ++++++ core/Subscriber.cxx | 501 ++++++++++++++++++++++++++++++ core/Subscriber.hpp | 98 ++++++ core/main.cxx | 249 +++++++++++++++ core/msg.hpp | 33 ++ gatectrl/MsgHandler.cxx | 3 + gatectrl/Publisher.cxx | 35 ++- gatectrl/Subscriber.cxx | 8 +- gatectrl/main.cxx | 5 +- httpserver/Publisher.cxx | 2 +- lib/System.hpp | 170 +++++++++++ lib/System.idl | 10 + lib/SystemCdrAux.hpp | 7 + lib/SystemCdrAux.ipp | 96 ++++++ lib/SystemPubSubTypes.cxx | 182 +++++++++++ lib/SystemPubSubTypes.hpp | 81 +++++ lib/SystemTypeObjectSupport.cxx | 97 ++++++ lib/SystemTypeObjectSupport.hpp | 13 + weigh/Publisher.cxx | 8 +- 21 files changed, 2186 insertions(+), 28 deletions(-) create mode 100644 core/Publisher.cxx create mode 100644 core/Publisher.hpp create mode 100644 core/Subscriber.cxx create mode 100644 core/Subscriber.hpp create mode 100644 core/main.cxx create mode 100644 core/msg.hpp diff --git a/CMakeLists.txt b/CMakeLists.txt index 6abcc31..343291b 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -9,10 +9,12 @@ set(CMAKE_CXX_EXTENSIONS OFF) set(fastcdr_DIR "/opt/fastdds/v3.2.2/lib/cmake/fastcdr") set(fastdds_DIR "/opt/fastdds/v3.2.2/share/fastdds/cmake") set(foonathan_memory_DIR "/opt/foonathan_memory/lib/foonathan_memory/cmake") +set(PahoMqttCpp_DIR "/opt/pahomqtt/v1.5.3/lib/cmake/PahoMqttCpp") # Find requirements find_package(fastcdr REQUIRED) find_package(fastdds 3 REQUIRED) +find_package(PahoMqttCpp REQUIRED) # Set CMAKE_BUILD_TYPE to Release by default. if(NOT CMAKE_BUILD_TYPE AND NOT CMAKE_CONFIGURATION_TYPES) @@ -41,6 +43,17 @@ target_link_libraries(System_lib fastcdr fastdds) # System_lib # ) +# Core Application. +add_executable(Core + core/main.cxx + core/Publisher.cxx + core/Subscriber.cxx + ) +target_include_directories(Core PRIVATE core) +target_link_libraries(Core fastcdr fastdds + System_lib PahoMqttCpp::paho-mqttpp3 + ) + # GateCtrl Application. add_executable(GateCtrl gatectrl/main.cxx diff --git a/core/Publisher.cxx b/core/Publisher.cxx new file mode 100644 index 0000000..c6fc75c --- /dev/null +++ b/core/Publisher.cxx @@ -0,0 +1,518 @@ +// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima). +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +/*! + * @file Publisher.cxx + * This file contains the implementation of the publisher functions. + * + * This file was generated by the tool fastddsgen. + */ + +#include "Publisher.hpp" + +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include + +#include "SystemPubSubTypes.hpp" +#include "json.hpp" +#include "msg.hpp" + +using namespace eprosima::fastdds::dds; +using json = nlohmann::json; + +PublisherApp::PublisherApp( + const int& domain_id) + : factory_(nullptr) + , participant_(nullptr) + , publisher_(nullptr) + , bar_topic_(nullptr) + , bar_writer_(nullptr) + , bar_type_(new WeighingSystem::BarCommandUpdatePubSubType()) + , light_topic_(nullptr) + , light_writer_(nullptr) + , light_type_(new WeighingSystem::LightsCommandUpdatePubSubType()) + , summary_topic_(nullptr) + , summary_writer_(nullptr) + , summary_type_(new WeighingSystem::SummaryUpdatePubSubType()) + , matched_(0) + , samples_sent_(0) + , stop_(false) +{ + // + + // Create the participant + DomainParticipantQos pqos = PARTICIPANT_QOS_DEFAULT; + pqos.name("Core_pub_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("Core Participant initialization failed"); + } + + // Register the type + bar_type_.register_type(participant_); + light_type_.register_type(participant_); + summary_type_.register_type(participant_); + + // Create the publisher + PublisherQos pub_qos = PUBLISHER_QOS_DEFAULT; + participant_->get_default_publisher_qos(pub_qos); + publisher_ = participant_->create_publisher(pub_qos, nullptr, StatusMask::none()); + if (publisher_ == nullptr) + { + throw std::runtime_error("Core Publisher initialization failed"); + } + + // Create the topic + TopicQos topic_qos = TOPIC_QOS_DEFAULT; + participant_->get_default_topic_qos(topic_qos); + bar_topic_ = participant_->create_topic("BarCommandUpdate", bar_type_.get_type_name(), topic_qos); + if (bar_topic_ == nullptr) + { + throw std::runtime_error("BarCommandUpdate Topic initialization failed"); + } + + // Create the data writer + DataWriterQos writer_qos = DATAWRITER_QOS_DEFAULT; + publisher_->get_default_datawriter_qos(writer_qos); + writer_qos.reliability().kind = ReliabilityQosPolicyKind::RELIABLE_RELIABILITY_QOS; + writer_qos.reliability().max_blocking_time = Duration_t(1, 0); + writer_qos.durability().kind = DurabilityQosPolicyKind::VOLATILE_DURABILITY_QOS; + writer_qos.history().kind = HistoryQosPolicyKind::KEEP_LAST_HISTORY_QOS; + writer_qos.history().depth = 1; + writer_qos.resource_limits().max_samples = 200; + writer_qos.resource_limits().max_instances = 1; + writer_qos.resource_limits().max_samples_per_instance = 100; + writer_qos.data_sharing().off(); + bar_writer_ = publisher_->create_datawriter(bar_topic_, writer_qos, this, StatusMask::all()); + if (bar_writer_ == nullptr) + { + throw std::runtime_error("WeighingSystem::BarUpdate DataWriter initialization failed"); + } + + // Create the topic + topic_qos = TOPIC_QOS_DEFAULT; + participant_->get_default_topic_qos(topic_qos); + light_topic_ = participant_->create_topic("LightsCommandUpdate", light_type_.get_type_name(), topic_qos); + if (light_topic_ == nullptr) + { + throw std::runtime_error("LightsCommandUpdate Topic initialization failed"); + } + + // Create the data writer + writer_qos = DATAWRITER_QOS_DEFAULT; + publisher_->get_default_datawriter_qos(writer_qos); + writer_qos.reliability().kind = ReliabilityQosPolicyKind::RELIABLE_RELIABILITY_QOS; + writer_qos.reliability().max_blocking_time = Duration_t(1, 0); + writer_qos.durability().kind = DurabilityQosPolicyKind::VOLATILE_DURABILITY_QOS; + writer_qos.history().kind = HistoryQosPolicyKind::KEEP_LAST_HISTORY_QOS; + writer_qos.history().depth = 1; + writer_qos.resource_limits().max_samples = 200; + writer_qos.resource_limits().max_instances = 1; + writer_qos.resource_limits().max_samples_per_instance = 100; + writer_qos.data_sharing().off(); + light_writer_ = publisher_->create_datawriter(light_topic_, writer_qos, this, StatusMask::all()); + if (light_writer_ == nullptr) + { + throw std::runtime_error("WeighingSystem::LightsUpdate DataWriter initialization failed"); + } + + // Create the topic + topic_qos = TOPIC_QOS_DEFAULT; + participant_->get_default_topic_qos(topic_qos); + summary_topic_ = participant_->create_topic("SummaryUpdate", summary_type_.get_type_name(), topic_qos); + if (summary_topic_ == nullptr) + { + throw std::runtime_error("SummaryUpdate Topic initialization failed"); + } + + // Create the data writer + writer_qos = DATAWRITER_QOS_DEFAULT; + publisher_->get_default_datawriter_qos(writer_qos); + writer_qos.reliability().kind = ReliabilityQosPolicyKind::RELIABLE_RELIABILITY_QOS; + writer_qos.reliability().max_blocking_time = Duration_t(1, 0); + writer_qos.durability().kind = DurabilityQosPolicyKind::TRANSIENT_LOCAL_DURABILITY_QOS; + writer_qos.history().kind = HistoryQosPolicyKind::KEEP_LAST_HISTORY_QOS; + writer_qos.history().depth = 1; + writer_qos.resource_limits().max_samples = 200; + writer_qos.resource_limits().max_instances = 1; + writer_qos.resource_limits().max_samples_per_instance = 100; + writer_qos.data_sharing().off(); + summary_writer_ = publisher_->create_datawriter(summary_topic_, writer_qos, this, StatusMask::all()); + if (summary_writer_ == nullptr) + { + throw std::runtime_error("WeighingSystem::SummaryUpdate DataWriter initialization failed"); + } +} + +PublisherApp::~PublisherApp() +{ + if (nullptr != participant_) + { + // Delete DDS entities contained within the DomainParticipant + participant_->delete_contained_entities(); + + // Delete DomainParticipant + factory_->delete_participant(participant_); + } +} + +void PublisherApp::on_publication_matched( + DataWriter* writer, + const PublicationMatchedStatus& info) +{ + if (info.current_count_change == 1) + { + { + std::lock_guard lock(mutex_); + matched_ = info.current_count; + } + std::cout << writer->get_topic()->get_name() << " Publisher matched." << std::endl; + cv_.notify_one(); + } + else if (info.current_count_change == -1) + { + { + std::lock_guard lock(mutex_); + matched_ = info.current_count; + } + std::cout << writer->get_topic()->get_name() << " Publisher unmatched." << std::endl; + } + else + { + std::cout << info.current_count_change + << " is not a valid value for PublicationMatchedStatus current count change" << std::endl; + } +} + +bool PublisherApp::bar_ctrl(const std::map& ctrl) +{ + WeighingSystem::BarCommandUpdate cmd; + + for (const auto &[port, opt] : ctrl) + { + if (port == "FrontBar") + { + if (opt == "up") + { + cmd.FrontBarSignalUp() = 1; + cmd.FrontBarSignalDown() = 0; + } + else + { + cmd.FrontBarSignalUp() = 0; + cmd.FrontBarSignalDown() = 1; + } + cmd.FrontBarEnable() = 1; + } + else if (port == "BackBar") + { + if (opt == "up") + { + cmd.BackBarSignalUp() = 1; + cmd.BackBarSignalDown() = 0; + } + else + { + cmd.BackBarSignalUp() = 0; + cmd.BackBarSignalDown() = 1; + } + cmd.BackBarEnable() = 1; + } + } + + std::cout << "send bar ctrl cmd" << std::endl; + bar_writer_->write(&cmd); + + return true; +} + +bool PublisherApp::light_ctrl(const std::map& ctrl) +{ + WeighingSystem::LightsCommandUpdate cmd; + + for (const auto &[port, opt] : ctrl) + { + if (port == "FrontLED") + { + if (opt == "red") + { + cmd.FrontLEDSignal() = 1; + } + else + { + cmd.FrontLEDSignal() = 2; + } + cmd.FrontLEDEnable() = 1; + } + else if (port == "BackLED") + { + if (opt == "red") + { + cmd.BackLEDSignal() = 1; + } + else + { + cmd.BackLEDSignal() = 2; + } + cmd.BackLEDEnable() = 1; + } + } + + std::cout << "send light ctrl cmd" << std::endl; + light_writer_->write(&cmd); + + return true; +} + +void PublisherApp::run(std::shared_ptr dev) +{ + uint8_t send_sum = 0; + std::string license_no = ""; + float stable_weight = 0; + + while (!is_stopped()) + { + //dds msg + std::unique_lock dds_lock(DdsMsgData::queue_cv_mtx_, std::try_to_lock); + if (dds_lock.owns_lock()) + { + if (!DdsMsgData::LicenseSnapUpdate_queue_.empty()) + { + WeighingSystem::LicenseSnapUpdate info = std::move(DdsMsgData::LicenseSnapUpdate_queue_.front()); + DdsMsgData::LicenseSnapUpdate_queue_.pop(); + dds_lock.unlock(); + + std::cout << "车牌: " << info.License() << std::endl; + std::cout << "类型: " << info.NewTag() << std::endl; + + if (license_no == "") + { + license_no = info.NewTag(); + } + } + else if (!DdsMsgData::ScaleInfo_queue_.empty()) + { + WeighingSystem::ScaleInfo info = std::move(DdsMsgData::ScaleInfo_queue_.front()); + DdsMsgData::ScaleInfo_queue_.pop(); + dds_lock.unlock(); + + std::cout << "有车: " << info.HasVehicle() << std::endl; + std::cout << "稳定: " << info.WeightOK() << std::endl; + std::cout << "设备状态: " << info.State() << std::endl; + std::cout << "实时重量: " << info.Value() << std::endl; + std::cout << "稳定重量: " << info.StableValue() << std::endl; + + if (info.WeightOK() == 1) + { + stable_weight = info.StableValue(); + } + else + { + stable_weight = 0; + send_sum = 0; + } + + if (info.HasVehicle() == 0) + { + license_no = ""; + } + } + else if (!DdsMsgData::BarUpdate_queue_.empty()) + { + WeighingSystem::BarUpdate info = std::move(DdsMsgData::BarUpdate_queue_.front()); + DdsMsgData::BarUpdate_queue_.pop(); + dds_lock.unlock(); + + std::cout << "前拦车器状态: " << info.FrontBarState() << std::endl; + std::cout << "后拦车器状态: " << info.BackBarState() << std::endl; + + if (dev != nullptr) + { + if (dev->is_connected()) + { + json j; + j["bar_state"]["FrontBar"] = info.FrontBarState(); + j["bar_state"]["BackBar"] = info.BackBarState(); + + auto mqtt_msg = mqtt::make_message("response", j.dump()); + mqtt_msg->set_qos(0); + mqtt_msg->set_retained(false); + dev->publish(mqtt_msg); + } + } + } + else if (!DdsMsgData::LightsUpdate_queue_.empty()) + { + WeighingSystem::LightsUpdate info = std::move(DdsMsgData::LightsUpdate_queue_.front()); + DdsMsgData::LightsUpdate_queue_.pop(); + dds_lock.unlock(); + + std::cout << "前红绿灯状态: " << info.FrontLEDState() << std::endl; + std::cout << "后红绿灯状态: " << info.BackLEDState() << std::endl; + + if (dev != nullptr) + { + if (dev->is_connected()) + { + json j; + j["light_state"]["FrontLED"] = info.FrontLEDState(); + j["light_state"]["BackLED"] = info.FrontLEDState(); + + auto mqtt_msg = mqtt::make_message("response", j.dump()); + mqtt_msg->set_qos(0); + mqtt_msg->set_retained(false); + dev->publish(mqtt_msg); + } + } + } + else if (!DdsMsgData::InfraredUpdate_queue_.empty()) + { + WeighingSystem::InfraredUpdate info = std::move(DdsMsgData::InfraredUpdate_queue_.front()); + DdsMsgData::InfraredUpdate_queue_.pop(); + dds_lock.unlock(); + + std::cout << "前红外对射状态: " << info.FrontResistanceSignal() << std::endl; + std::cout << "后红外对射状态: " << info.BackResistanceSignal() << std::endl; + + if (dev != nullptr) + { + if (dev->is_connected()) + { + json j; + j["infrared_state"]["FrontResistance"] = info.FrontResistanceSignal(); + j["infrared_state"]["BackResistance"] = info.BackResistanceSignal(); + + auto mqtt_msg = mqtt::make_message("response", j.dump()); + mqtt_msg->set_qos(0); + mqtt_msg->set_retained(false); + dev->publish(mqtt_msg); + } + } + } + else if (!DdsMsgData::WeightInfoOk_queue_.empty()) + { + WeighingSystem::WeightInfoOk info = std::move(DdsMsgData::WeightInfoOk_queue_.front()); + DdsMsgData::WeightInfoOk_queue_.pop(); + dds_lock.unlock(); + } + else if (!DdsMsgData::WeightInfoError_queue_.empty()) + { + WeighingSystem::WeightInfoError info = std::move(DdsMsgData::WeightInfoError_queue_.front()); + DdsMsgData::WeightInfoError_queue_.pop(); + dds_lock.unlock(); + + send_sum = 0; + } + else + { + dds_lock.unlock(); + } + } + + // mqtt msg + std::unique_lock mqtt_lock(MqttMsgData::queue_cv_mtx_, std::try_to_lock); + if (mqtt_lock.owns_lock()) + { + if (!MqttMsgData::Mqtt_msg_queue_.empty()) + { + mqtt::const_message_ptr msg = std::move(MqttMsgData::Mqtt_msg_queue_.front()); + MqttMsgData::Mqtt_msg_queue_.pop(); + mqtt_lock.unlock(); + + json cmd = json::parse(msg->to_string()); + + if (cmd.contains("bar_ctrl")) + { + std::map ctrl = cmd["bar_ctrl"].get>(); + bar_ctrl(ctrl); + } + else if (cmd.contains("light_ctrl")) + { + std::map ctrl = cmd["light_ctrl"].get>(); + light_ctrl(ctrl); + } + } + else + { + mqtt_lock.unlock(); + } + } + + // logic + if (send_sum == 0) + { + if ((stable_weight != 0) && (license_no != "")) + { + std::cout << "send summary" << std::endl; + WeighingSystem::SummaryUpdate info; + info.License() = license_no; + info.StableValue() = stable_weight; + summary_writer_->write(&info); + send_sum = 1; + } + } + + // Wait for period or stop event + std::unique_lock period_lock(mutex_); + cv_.wait_for(period_lock, std::chrono::milliseconds(period_ms_), [this]() + { + return is_stopped(); + }); + } +} + + +bool PublisherApp::publish() +{ + bool ret = false; + // Wait for the data endpoints discovery + std::unique_lock matched_lock(mutex_); + cv_.wait(matched_lock, [&]() + { + // at least one has been discovered + return ((matched_ > 0) || is_stopped()); + }); + + if (!is_stopped()) + { + /* Initialize your structure here */ + WeighingSystem::BarCommandUpdate sample_; + ret = (RETCODE_OK == bar_writer_->write(&sample_)); + } + return ret; +} + +bool PublisherApp::is_stopped() +{ + return stop_.load(); +} + +void PublisherApp::stop() +{ + stop_.store(true); + cv_.notify_one(); +} \ No newline at end of file diff --git a/core/Publisher.hpp b/core/Publisher.hpp new file mode 100644 index 0000000..173d78b --- /dev/null +++ b/core/Publisher.hpp @@ -0,0 +1,85 @@ +// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima). +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +/*! + * @file PublisherApp.hpp + * This header file contains the declaration of the publisher functions. + * + * This file was generated by the tool fastddsgen. + */ + +#ifndef FAST_DDS_GENERATED__PUBLISHERAPP_HPP +#define FAST_DDS_GENERATED__PUBLISHERAPP_HPP + +#include + +#include +#include +#include +#include + +#include "mqtt/async_client.h" + +class PublisherApp : public eprosima::fastdds::dds::DataWriterListener +{ +public: + + PublisherApp( + const int& domain_id); + + ~PublisherApp(); + + //! Publisher matched method + void on_publication_matched( + eprosima::fastdds::dds::DataWriter* writer, + const eprosima::fastdds::dds::PublicationMatchedStatus& info) override; + + //! Run publisher + void run(std::shared_ptr dev); + + //! Trigger the end of execution + void stop(); + + bool bar_ctrl(const std::map& ctrl); + bool light_ctrl(const std::map& ctrl); + +private: + + //! Return the current state of execution + bool is_stopped(); + + //! Publish a sample + bool publish(); + + std::shared_ptr factory_; + eprosima::fastdds::dds::DomainParticipant* participant_; + eprosima::fastdds::dds::Publisher* publisher_; + eprosima::fastdds::dds::Topic* bar_topic_; + eprosima::fastdds::dds::DataWriter* bar_writer_; + eprosima::fastdds::dds::TypeSupport bar_type_; + eprosima::fastdds::dds::Topic* light_topic_; + eprosima::fastdds::dds::DataWriter* light_writer_; + eprosima::fastdds::dds::TypeSupport light_type_; + eprosima::fastdds::dds::Topic* summary_topic_; + eprosima::fastdds::dds::DataWriter* summary_writer_; + eprosima::fastdds::dds::TypeSupport summary_type_; + std::condition_variable cv_; + int32_t matched_; + std::mutex mutex_; + const uint32_t period_ms_ = 100; // in ms + uint16_t samples_sent_; + std::atomic stop_; +}; + +#endif // FAST_DDS_GENERATED__PUBLISHERAPP_HPP \ No newline at end of file diff --git a/core/Subscriber.cxx b/core/Subscriber.cxx new file mode 100644 index 0000000..f57283c --- /dev/null +++ b/core/Subscriber.cxx @@ -0,0 +1,501 @@ +// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima). +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +/*! + * @file Subscriber.cxx + * This file contains the implementation of the subscriber functions. + * + * This file was generated by the tool fastddsgen. + */ + +#include "Subscriber.hpp" + +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +#include "SystemPubSubTypes.hpp" + +#include "msg.hpp" + +using namespace eprosima::fastdds::dds; + +SubscriberApp::SubscriberApp( + const int& domain_id) + : factory_(nullptr) + , participant_(nullptr) + , subscriber_(nullptr) + , bar_topic_(nullptr) + , bar_reader_(nullptr) + , bar_type_(new WeighingSystem::BarUpdatePubSubType()) + , light_topic_(nullptr) + , light_reader_(nullptr) + , light_type_(new WeighingSystem::LightsUpdatePubSubType()) + , infraredinfo_topic_(nullptr) + , infraredinfo_reader_(nullptr) + , infraredinfo_type_(new WeighingSystem::InfraredUpdatePubSubType()) + , infraredcommand_topic_(nullptr) + , infraredcommand_reader_(nullptr) + , infraredcommand_type_(new WeighingSystem::InfraredCommandUpdatePubSubType()) + , license_topic_(nullptr) + , license_reader_(nullptr) + , license_type_(new WeighingSystem::LicenseSnapUpdatePubSubType()) + , scaleinfo_topic_(nullptr) + , scaleinfo_reader_(nullptr) + , scaleinfo_type_(new WeighingSystem::ScaleInfoPubSubType()) + , weightinfook_topic_(nullptr) + , weightinfook_reader_(nullptr) + , weightinfook_type_(new WeighingSystem::WeightInfoOkPubSubType()) + , weightinfoerror_topic_(nullptr) + , weightinfoerror_reader_(nullptr) + , weightinfoerror_type_(new WeighingSystem::WeightInfoErrorPubSubType()) + , samples_received_(0) + , stop_(false) +{ + // Create the participant + DomainParticipantQos pqos = PARTICIPANT_QOS_DEFAULT; + pqos.name("WeighingSystem_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("WeighingSystem Participant initialization failed"); + } + + // Register the type + bar_type_.register_type(participant_); + light_type_.register_type(participant_); + infraredinfo_type_.register_type(participant_); + infraredcommand_type_.register_type(participant_); + license_type_.register_type(participant_); + scaleinfo_type_.register_type(participant_); + weightinfook_type_.register_type(participant_); + weightinfoerror_type_.register_type(participant_); + + // Create the subscriber + 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("WeighingSystem Subscriber initialization failed"); + } + + // Create the topic + TopicQos topic_qos = TOPIC_QOS_DEFAULT; + participant_->get_default_topic_qos(topic_qos); + bar_topic_ = participant_->create_topic("BarUpdate", bar_type_.get_type_name(), topic_qos); + if (bar_topic_ == nullptr) + { + throw std::runtime_error("WeighingSystem::BarUpdate Topic initialization failed"); + } + + // Create the reader + 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::TRANSIENT_LOCAL_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(); + bar_reader_ = subscriber_->create_datareader(bar_topic_, reader_qos, this, StatusMask::all()); + if (bar_reader_ == nullptr) + { + throw std::runtime_error("WeighingSystem::BarUpdate DataReader initialization failed"); + } + + // Create the topic + topic_qos = TOPIC_QOS_DEFAULT; + participant_->get_default_topic_qos(topic_qos); + light_topic_ = participant_->create_topic("LightsUpdate", light_type_.get_type_name(), topic_qos); + if (light_topic_ == nullptr) + { + throw std::runtime_error("WeighingSystem::LightsUpdate Topic initialization failed"); + } + + // Create the reader + 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::TRANSIENT_LOCAL_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(); + light_reader_ = subscriber_->create_datareader(light_topic_, reader_qos, this, StatusMask::all()); + if (light_reader_ == nullptr) + { + throw std::runtime_error("WeighingSystem::LightsUpdate DataReader initialization failed"); + } + + // Create the topic + topic_qos = TOPIC_QOS_DEFAULT; + participant_->get_default_topic_qos(topic_qos); + infraredinfo_topic_ = participant_->create_topic("InfraredUpdate", infraredinfo_type_.get_type_name(), topic_qos); + if (infraredinfo_topic_ == nullptr) + { + throw std::runtime_error("WeighingSystem::InfraredUpdate Topic initialization failed"); + } + + // Create the reader + 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::TRANSIENT_LOCAL_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(); + infraredinfo_reader_ = subscriber_->create_datareader(infraredinfo_topic_, reader_qos, this, StatusMask::all()); + if (infraredinfo_reader_ == nullptr) + { + throw std::runtime_error("WeighingSystem::InfraredUpdate DataReader initialization failed"); + } + + // Create the topic + topic_qos = TOPIC_QOS_DEFAULT; + participant_->get_default_topic_qos(topic_qos); + infraredcommand_topic_ = participant_->create_topic("InfraredCommandUpdate", infraredcommand_type_.get_type_name(), topic_qos); + if (infraredcommand_topic_ == nullptr) + { + throw std::runtime_error("WeighingSystem::InfraredCommandUpdate Topic initialization failed"); + } + + // Create the reader + 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::TRANSIENT_LOCAL_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(); + infraredcommand_reader_ = subscriber_->create_datareader(infraredcommand_topic_, reader_qos, this, StatusMask::all()); + if (infraredcommand_reader_ == nullptr) + { + throw std::runtime_error("WeighingSystem::InfraredCommandUpdate DataReader initialization failed"); + } + + // Create the topic + topic_qos = TOPIC_QOS_DEFAULT; + participant_->get_default_topic_qos(topic_qos); + license_topic_ = participant_->create_topic("LicenseSnapUpdate", license_type_.get_type_name(), topic_qos); + if (license_topic_ == nullptr) + { + throw std::runtime_error("WeighingSystem::LicenseSnapUpdate Topic initialization failed"); + } + + // Create the reader + 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(); + license_reader_ = subscriber_->create_datareader(license_topic_, reader_qos, this, StatusMask::all()); + if (license_reader_ == nullptr) + { + throw std::runtime_error("WeighingSystem::LicenseSnapUpdate DataReader initialization failed"); + } + + // Create the topic + topic_qos = TOPIC_QOS_DEFAULT; + participant_->get_default_topic_qos(topic_qos); + scaleinfo_topic_ = participant_->create_topic("ScaleInfo", scaleinfo_type_.get_type_name(), topic_qos); + if (scaleinfo_topic_ == nullptr) + { + throw std::runtime_error("WeighingSystem::ScaleInfo Topic initialization failed"); + } + + // Create the reader + 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(); + scaleinfo_reader_ = subscriber_->create_datareader(scaleinfo_topic_, reader_qos, this, StatusMask::all()); + if (scaleinfo_reader_ == nullptr) + { + throw std::runtime_error("WeighingSystem::ScaleInfo DataReader initialization failed"); + } + + // Create the topic + 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("WeighingSystem::WeightInfoOk Topic initialization failed"); + } + + // Create the reader + 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::TRANSIENT_LOCAL_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("WeighingSystem::WeightInfoOk DataReader initialization failed"); + } + + // Create the topic + topic_qos = TOPIC_QOS_DEFAULT; + participant_->get_default_topic_qos(topic_qos); + weightinfoerror_topic_ = participant_->create_topic("WeightInfoError", weightinfoerror_type_.get_type_name(), topic_qos); + if (weightinfoerror_topic_ == nullptr) + { + throw std::runtime_error("WeighingSystem::WeightInfoError Topic initialization failed"); + } + + // Create the reader + 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::TRANSIENT_LOCAL_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(); + weightinfoerror_reader_ = subscriber_->create_datareader(weightinfoerror_topic_, reader_qos, this, StatusMask::all()); + if (weightinfoerror_reader_ == nullptr) + { + throw std::runtime_error("WeighingSystem::WeightInfoError DataReader initialization failed"); + } +} + +SubscriberApp::~SubscriberApp() +{ + if (nullptr != participant_) + { + // Delete DDS entities contained within the DomainParticipant + participant_->delete_contained_entities(); + + // Delete DomainParticipant + factory_->delete_participant(participant_); + } +} + +void SubscriberApp::on_subscription_matched( + DataReader* reader, + const SubscriptionMatchedStatus& info) +{ + if (info.current_count_change == 1) + { + std::cout << reader->get_topicdescription()->get_name() << " Subscriber matched." << std::endl; + } + else if (info.current_count_change == -1) + { + std::cout << reader->get_topicdescription()->get_name() << " Subscriber unmatched." << std::endl; + } + else + { + std::cout << info.current_count_change + << " is not a valid value for SubscriptionMatchedStatus current count change" << std::endl; + } +} + +void SubscriberApp::on_data_available( + DataReader* reader) +{ + SampleInfo info; + std::string topic_name = reader->get_topicdescription()->get_name(); + std::cout << topic_name << std::endl; + + if (topic_name == "BarUpdate") + { + WeighingSystem::BarUpdate 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::BarUpdate_queue_.push(std::move(sample_)); + lock.unlock(); + } + } + } + } + else if (topic_name == "LightsUpdate") + { + WeighingSystem::LightsUpdate 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::LightsUpdate_queue_.push(std::move(sample_)); + lock.unlock(); + } + } + } + } + else if (topic_name == "InfraredUpdate") + { + WeighingSystem::InfraredUpdate 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::InfraredUpdate_queue_.push(std::move(sample_)); + lock.unlock(); + } + } + } + } + else if (topic_name == "InfraredCommandUpdate") + { + WeighingSystem::InfraredCommandUpdate 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::InfraredCommandUpdate_queue_.push(std::move(sample_)); + lock.unlock(); + } + } + } + } + else if (topic_name == "LicenseSnapUpdate") + { + WeighingSystem::LicenseSnapUpdate 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::LicenseSnapUpdate_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(); + } + } + } + } + else 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(); + } + } + } + } +} + +void SubscriberApp::run(std::shared_ptr dev) +{ + 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 SubscriberApp::is_stopped() +{ + return stop_.load(); +} + +void SubscriberApp::stop() +{ + stop_.store(true); + terminate_cv_.notify_all(); +} \ No newline at end of file diff --git a/core/Subscriber.hpp b/core/Subscriber.hpp new file mode 100644 index 0000000..d4b6db2 --- /dev/null +++ b/core/Subscriber.hpp @@ -0,0 +1,98 @@ +// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima). +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +/*! + * @file SubscriberApp.hpp + * This header file contains the declaration of the subscriber functions. + * + * This file was generated by the tool fastddsgen. + */ + +#ifndef FAST_DDS_GENERATED__SUBSCRIBERAPP_HPP +#define FAST_DDS_GENERATED__SUBSCRIBERAPP_HPP + +#include + +#include +#include +#include +#include + +#include "System.hpp" +#include "mqtt/async_client.h" + +class SubscriberApp : public eprosima::fastdds::dds::DataReaderListener +{ +public: + + SubscriberApp( + const int& domain_id); + + virtual ~SubscriberApp(); + + //! Subscription callback + void on_data_available( + eprosima::fastdds::dds::DataReader* reader) override; + + //! Subscriber matched method + void on_subscription_matched( + eprosima::fastdds::dds::DataReader* reader, + const eprosima::fastdds::dds::SubscriptionMatchedStatus& info) override; + + //! Run subscriber + void run(std::shared_ptr dev); + + //! Trigger the end of execution + void stop(); + +private: + + //! Return the current state of execution + bool is_stopped(); + + std::shared_ptr factory_; + eprosima::fastdds::dds::DomainParticipant* participant_; + eprosima::fastdds::dds::Subscriber* subscriber_; + eprosima::fastdds::dds::Topic* bar_topic_; + eprosima::fastdds::dds::DataReader* bar_reader_; + eprosima::fastdds::dds::TypeSupport bar_type_; + eprosima::fastdds::dds::Topic* light_topic_; + eprosima::fastdds::dds::DataReader* light_reader_; + eprosima::fastdds::dds::TypeSupport light_type_; + eprosima::fastdds::dds::Topic* infraredinfo_topic_; + eprosima::fastdds::dds::DataReader* infraredinfo_reader_; + eprosima::fastdds::dds::TypeSupport infraredinfo_type_; + eprosima::fastdds::dds::Topic* infraredcommand_topic_; + eprosima::fastdds::dds::DataReader* infraredcommand_reader_; + eprosima::fastdds::dds::TypeSupport infraredcommand_type_; + eprosima::fastdds::dds::Topic* license_topic_; + eprosima::fastdds::dds::DataReader* license_reader_; + eprosima::fastdds::dds::TypeSupport license_type_; + eprosima::fastdds::dds::Topic* scaleinfo_topic_; + eprosima::fastdds::dds::DataReader* scaleinfo_reader_; + eprosima::fastdds::dds::TypeSupport scaleinfo_type_; + eprosima::fastdds::dds::Topic* weightinfook_topic_; + eprosima::fastdds::dds::DataReader* weightinfook_reader_; + eprosima::fastdds::dds::TypeSupport weightinfook_type_; + eprosima::fastdds::dds::Topic* weightinfoerror_topic_; + eprosima::fastdds::dds::DataReader* weightinfoerror_reader_; + eprosima::fastdds::dds::TypeSupport weightinfoerror_type_; + uint16_t samples_received_; + std::atomic stop_; + uint32_t period_ms_ = 100; // in ms + mutable std::mutex terminate_cv_mtx_; + std::condition_variable terminate_cv_; +}; + +#endif // FAST_DDS_GENERATED__SUBSCRIBERAPP_HPP \ No newline at end of file diff --git a/core/main.cxx b/core/main.cxx new file mode 100644 index 0000000..20c3928 --- /dev/null +++ b/core/main.cxx @@ -0,0 +1,249 @@ +// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima). +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +/*! + * @file main.cxx + * This file acts as a main entry point to the application. + * + * This file was generated by the tool fastddsgen. + */ + +#include +#include +#include +#include +#include +#include + +#include + +#include "Subscriber.hpp" +#include "Publisher.hpp" + +#include "mqtt/async_client.h" +#include "msg.hpp" + +#define VERSION "v1.0" + +using eprosima::fastdds::dds::Log; + +std::function stop_handler; +void signal_handler( + int signum) +{ + stop_handler(signum); +} + +std::string parse_signal( + const int& signum) +{ + switch (signum) + { + case SIGINT: + return "SIGINT"; + case SIGTERM: + return "SIGTERM"; +#ifndef _WIN32 + case SIGQUIT: + return "SIGQUIT"; + case SIGHUP: + return "SIGHUP"; +#endif // _WIN32 + default: + return "UNKNOWN SIGNAL"; + } +} + +class mqtt_callback : public mqtt::callback +{ +public: + mqtt_callback( + mqtt::async_client& cli, + std::vector topics, + std::vector qos) + : cli_(cli) + , sub_topic_(std::move(topics)) + , sub_qos_(std::move(qos)) + {} + + void connected(const std::string &cause) override + { + std::cout << "[MQTT] Connected: " << cause << std::endl; + for (size_t i = 0; i < sub_topic_.size(); ++i) + { + std::cout << "topic: " << sub_topic_[i] << std::endl; + std::cout << "qos: " << sub_qos_[i] << std::endl; + cli_.subscribe(sub_topic_[i], sub_qos_[i]); + } + } + + void connection_lost(const std::string &cause) override + { + std::cout << "[MQTT] Connection lost: " << cause << std::endl; + } + + void message_arrived(mqtt::const_message_ptr msg) override + { + std::cout << "[MQTT] " + << msg->get_topic() + << " -> " + << msg->to_string() + << std::endl; + { + std::unique_lock lock(MqttMsgData::queue_cv_mtx_); + MqttMsgData::Mqtt_msg_queue_.push(msg); + lock.unlock(); + } + } + + void delivery_complete(mqtt::delivery_token_ptr tok) override + { + // 发布消息完成 + } +private: + mqtt::async_client& cli_; + std::vector sub_topic_; + std::vector sub_qos_; +}; + +std::queue DdsMsgData::BarUpdate_queue_; +std::queue DdsMsgData::LightsUpdate_queue_; +std::queue DdsMsgData::InfraredUpdate_queue_; +std::queue DdsMsgData::InfraredCommandUpdate_queue_; +std::queue DdsMsgData::LicenseSnapUpdate_queue_; +std::queue DdsMsgData::ScaleInfo_queue_; +std::queue DdsMsgData::WeightInfoOk_queue_; +std::queue DdsMsgData::WeightInfoError_queue_; +std::mutex DdsMsgData::queue_cv_mtx_; +std::queue MqttMsgData::Mqtt_msg_queue_; +std::mutex MqttMsgData::queue_cv_mtx_; +std::map SubPubData::sub_to_pub_queue_; +std::mutex SubPubData::queue_cv_mtx_; + +int main(int argc, char** argv) +{ + auto ret = EXIT_SUCCESS; + std::shared_ptr sub; + std::shared_ptr pub; + std::shared_ptr dev = nullptr; + + int domain_id = 0; + std::string mqtt_server; + std::string mqtt_username = "admin"; + std::string mqtt_password = "admin"; + std::string mqtt_id = "core_0"; + const std::vector topic = {"command"}; + const std::vector qos = {0}; + + for (int i = 1; i < argc; i++) + { + if (strcmp(argv[i], "--domain") == 0 && i + 1 < argc) + { + domain_id = atoi(argv[++i]); + mqtt_id = std::string("core_") + std::string(argv[i]); + } + else if (strcmp(argv[i], "--addr") == 0 && i + 1 < argc) + { + mqtt_server = argv[++i]; + } + else if (strcmp(argv[i], "--user") == 0 && i + 1 < argc) + { + mqtt_username = argv[++i]; + } + else if (strcmp(argv[i], "--password") == 0 && i + 1 < argc) + { + mqtt_password = argv[++i]; + } + else if (strcmp(argv[i], "--version") == 0) + { + std::cout << "Vesrion: " << VERSION << "\n"; + return EXIT_SUCCESS; + } + else if (strcmp(argv[i], "--help") == 0) + { + std::cout << "Usage: [options]\n" + << "Options:\n" + << " --domain Set domain ID\n" + << " --addr Set mqtt addr (e.g., mqtt://192.168.1.1:1883)\n" + << " --user Set mqtt username (e.g., admin)\n" + << " --password Set mqtt password (e.g., admin)\n" + << " --version Show software Version\n" + << " --help Show this help message\n"; + return EXIT_SUCCESS; + } + else + { + std::cerr << "Unknown option: " << argv[i] << "\n"; + std::cout << "Usage: [options]\n" + << "Options:\n" + << " --domain Set domain ID\n" + << " --addr Set mqtt addr (e.g., mqtt://192.168.1.1:1883)\n" + << " --user Set mqtt username (e.g., admin)\n" + << " --password Set mqtt password (e.g., admin)\n" + << " --version Show software Version\n" + << " --help Show this help message\n"; + return EXIT_FAILURE; + } + } + + sub = std::make_shared(domain_id); + pub = std::make_shared(domain_id); + + std::shared_ptr cb; + if (mqtt_server != "") + { + dev = std::make_shared(mqtt_server, mqtt_id); + + auto connOpts = mqtt::connect_options_builder::v3() + .user_name(mqtt_username) + .password(mqtt_password) + .keep_alive_interval(std::chrono::seconds(30)) + .automatic_reconnect(std::chrono::seconds(2), std::chrono::seconds(30)) + .clean_session(false) + .finalize(); + + cb = std::make_shared(*dev, topic, qos); + dev->set_callback(*cb); + + std::cout << "Connecting to broker..." << std::endl; + dev->connect(connOpts); + std::cout << "MQTT running..." << std::endl; + } + + std::thread sub_thread(&SubscriberApp::run, sub, dev); + std::thread pub_thread(&PublisherApp::run, pub, dev); + + std::cout << "Program is running. Please press Ctrl+C to stop at any time." << std::endl; + + stop_handler = [&](int signum) + { + std::cout << "\n" << parse_signal(signum) << " received, stopping " << argv[1] + << " execution." << std::endl; + sub->stop(); + pub->stop(); + }; + + signal(SIGINT, signal_handler); + signal(SIGTERM, signal_handler); +#ifndef _WIN32 + signal(SIGQUIT, signal_handler); + signal(SIGHUP, signal_handler); +#endif // _WIN32 + + sub_thread.join(); + pub_thread.join(); + + Log::Reset(); + return ret; +} diff --git a/core/msg.hpp b/core/msg.hpp new file mode 100644 index 0000000..87d5800 --- /dev/null +++ b/core/msg.hpp @@ -0,0 +1,33 @@ +#ifndef _MSG_HPP_ +#define _MSG_HPP_ + +#include +#include +#include "System.hpp" + +class DdsMsgData { +public: + static std::queue BarUpdate_queue_; + static std::queue LightsUpdate_queue_; + static std::queue InfraredUpdate_queue_; + static std::queue InfraredCommandUpdate_queue_; + static std::queue LicenseSnapUpdate_queue_; + static std::queue ScaleInfo_queue_; + static std::queue WeightInfoOk_queue_; + static std::queue WeightInfoError_queue_; + static std::mutex queue_cv_mtx_; +}; + +class MqttMsgData { +public: + static std::queue Mqtt_msg_queue_; + static std::mutex queue_cv_mtx_; +}; + +class SubPubData { +public: + static std::map sub_to_pub_queue_; + static std::mutex queue_cv_mtx_; +}; + +#endif \ No newline at end of file diff --git a/gatectrl/MsgHandler.cxx b/gatectrl/MsgHandler.cxx index 0c99d5b..bc21dba 100644 --- a/gatectrl/MsgHandler.cxx +++ b/gatectrl/MsgHandler.cxx @@ -26,6 +26,8 @@ MsgHandler::MsgHandler() : fd(-1) this->new_bar_state.BackResistanceSignal = UNBLOCK; this->old_bar_state = this->new_bar_state; + this->old_bar_state.FrontResistanceSignal = BLOCK; + this->old_bar_state.BackResistanceSignal = BLOCK; } void MsgHandler::HandleDdsMsg(const std::map& msg) @@ -224,6 +226,7 @@ int MsgHandler::ParseDeviceMsg(std::vector& data) break; ret = 0; case 0x52: + std::cout << "gpio change" << std::endl; for (auto &p : this->iomap.imap) { p.second.second = data[p.second.first+2]; diff --git a/gatectrl/Publisher.cxx b/gatectrl/Publisher.cxx index e97c4d2..d639102 100644 --- a/gatectrl/Publisher.cxx +++ b/gatectrl/Publisher.cxx @@ -53,7 +53,7 @@ PublisherApp::PublisherApp( , light_type_(new WeighingSystem::LightsUpdatePubSubType()) , infrared_topic_(nullptr) , infrared_writer_(nullptr) - , infrared_type_(new WeighingSystem::LightsUpdatePubSubType()) + , infrared_type_(new WeighingSystem::InfraredUpdatePubSubType()) , matched_(0) , samples_sent_(0) , stop_(false) @@ -62,14 +62,14 @@ PublisherApp::PublisherApp( // Create the participant DomainParticipantQos pqos = PARTICIPANT_QOS_DEFAULT; - pqos.name("WeighingSystem_pub_participant"); + pqos.name("GateCtrl_pub_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("WeighingSystem Participant initialization failed"); + throw std::runtime_error("GateCtrl Participant initialization failed"); } // Register the type @@ -83,7 +83,7 @@ PublisherApp::PublisherApp( publisher_ = participant_->create_publisher(pub_qos, nullptr, StatusMask::none()); if (publisher_ == nullptr) { - throw std::runtime_error("WeighingSystem Publisher initialization failed"); + throw std::runtime_error("GateCtrl Publisher initialization failed"); } // Create the topic @@ -214,20 +214,7 @@ void PublisherApp::run(std::shared_ptr handler) { if(handler->isOpen() == true) { - if(handler->HandleDeviceMsg()) - { - // if ((handler->old_bar_state.FrontResistanceSignal != handler->new_bar_state.FrontResistanceSignal) || - // (handler->old_bar_state.BackResistanceSignal != handler->new_bar_state.BackResistanceSignal)) - // { - WeighingSystem::InfraredUpdate info; - info.FrontResistanceSignal() = handler->new_bar_state.FrontResistanceSignal; - info.BackResistanceSignal() = handler->new_bar_state.BackResistanceSignal; - handler->old_bar_state.FrontResistanceSignal = handler->new_bar_state.FrontResistanceSignal; - handler->old_bar_state.BackResistanceSignal = handler->new_bar_state.BackResistanceSignal; - - infrared_writer_->write(&info); - // } - } + handler->HandleDeviceMsg(); } if ((handler->old_bar_state.FrontBarState != handler->new_bar_state.FrontBarState) || \ @@ -256,6 +243,18 @@ void PublisherApp::run(std::shared_ptr handler) light_writer_->write(&info); } + if ((handler->old_bar_state.FrontResistanceSignal != handler->new_bar_state.FrontResistanceSignal) || + (handler->old_bar_state.BackResistanceSignal != handler->new_bar_state.BackResistanceSignal)) + { + WeighingSystem::InfraredUpdate info; + info.FrontResistanceSignal() = handler->new_bar_state.FrontResistanceSignal; + info.BackResistanceSignal() = handler->new_bar_state.BackResistanceSignal; + handler->old_bar_state.FrontResistanceSignal = handler->new_bar_state.FrontResistanceSignal; + handler->old_bar_state.BackResistanceSignal = handler->new_bar_state.BackResistanceSignal; + + infrared_writer_->write(&info); + } + // Wait for period or stop event std::unique_lock period_lock(mutex_); cv_.wait_for(period_lock, std::chrono::milliseconds(period_ms_), [this]() diff --git a/gatectrl/Subscriber.cxx b/gatectrl/Subscriber.cxx index b591b72..528b43d 100644 --- a/gatectrl/Subscriber.cxx +++ b/gatectrl/Subscriber.cxx @@ -54,14 +54,14 @@ SubscriberApp::SubscriberApp( { // Create the participant DomainParticipantQos pqos = PARTICIPANT_QOS_DEFAULT; - pqos.name("WeighingSystem_sub_participant"); + pqos.name("GateCtrl_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("WeighingSystem Participant initialization failed"); + throw std::runtime_error("GateCtrl Participant initialization failed"); } // Register the type @@ -74,7 +74,7 @@ SubscriberApp::SubscriberApp( subscriber_ = participant_->create_subscriber(sub_qos, nullptr, StatusMask::none()); if (subscriber_ == nullptr) { - throw std::runtime_error("WeighingSystem Subscriber initialization failed"); + throw std::runtime_error("GateCtrl Subscriber initialization failed"); } // Create the topic @@ -170,7 +170,7 @@ void SubscriberApp::on_data_available( std::string topic_name = reader->get_topicdescription()->get_name(); std::cout << topic_name << std::endl; - if (topic_name == "BarCommandUpdateTopic") + if (topic_name == "BarCommandUpdate") { WeighingSystem::BarCommandUpdate sample_; while ((!is_stopped()) && (RETCODE_OK == reader->take_next_sample(&sample_, &info))) diff --git a/gatectrl/main.cxx b/gatectrl/main.cxx index 514171c..2fdfa5f 100644 --- a/gatectrl/main.cxx +++ b/gatectrl/main.cxx @@ -80,7 +80,7 @@ int main(int argc, char** argv) const char* port = "can0"; const char* baudrate = "500000"; const char* device = "5serial"; - int device_id = 0; + int device_id = 1; for (int i = 1; i < argc; i++) { @@ -150,6 +150,7 @@ int main(int argc, char** argv) dev->device = device; dev->device_id = device_id; + std::thread sub_thread(&SubscriberApp::run, sub, dev); std::thread pub_thread(&PublisherApp::run, pub, dev); std::cout << "Program is running. Please press Ctrl+C to stop at any time." << std::endl; @@ -158,6 +159,7 @@ int main(int argc, char** argv) { std::cout << "\n" << parse_signal(signum) << " received, stopping " << argv[1] << " execution." << std::endl; + sub->stop(); pub->stop(); }; @@ -168,6 +170,7 @@ int main(int argc, char** argv) signal(SIGHUP, signal_handler); #endif // _WIN32 + sub_thread.join(); pub_thread.join(); Log::Reset(); diff --git a/httpserver/Publisher.cxx b/httpserver/Publisher.cxx index b341f8f..19cc26d 100644 --- a/httpserver/Publisher.cxx +++ b/httpserver/Publisher.cxx @@ -91,7 +91,7 @@ PublisherApp::PublisherApp( publisher_->get_default_datawriter_qos(writer_qos); writer_qos.reliability().kind = ReliabilityQosPolicyKind::RELIABLE_RELIABILITY_QOS; writer_qos.reliability().max_blocking_time = Duration_t(1, 0); - writer_qos.durability().kind = DurabilityQosPolicyKind::TRANSIENT_LOCAL_DURABILITY_QOS; + writer_qos.durability().kind = DurabilityQosPolicyKind::VOLATILE_DURABILITY_QOS; writer_qos.history().kind = HistoryQosPolicyKind::KEEP_LAST_HISTORY_QOS; writer_qos.history().depth = 1; writer_qos.resource_limits().max_samples = 200; diff --git a/lib/System.hpp b/lib/System.hpp index 6cea57f..fa1b062 100644 --- a/lib/System.hpp +++ b/lib/System.hpp @@ -54,6 +54,176 @@ namespace WeighingSystem { +/*! + * @brief This class represents the structure SummaryUpdate defined by the user in the IDL file. + * @ingroup System + */ +class SummaryUpdate +{ +public: + + /*! + * @brief Default constructor. + */ + eProsima_user_DllExport SummaryUpdate() + { + } + + /*! + * @brief Default destructor. + */ + eProsima_user_DllExport ~SummaryUpdate() + { + } + + /*! + * @brief Copy constructor. + * @param x Reference to the object SummaryUpdate that will be copied. + */ + eProsima_user_DllExport SummaryUpdate( + const SummaryUpdate& x) + { + m_License = x.m_License; + + m_StableValue = x.m_StableValue; + + } + + /*! + * @brief Move constructor. + * @param x Reference to the object SummaryUpdate that will be copied. + */ + eProsima_user_DllExport SummaryUpdate( + SummaryUpdate&& x) noexcept + { + m_License = std::move(x.m_License); + m_StableValue = x.m_StableValue; + } + + /*! + * @brief Copy assignment. + * @param x Reference to the object SummaryUpdate that will be copied. + */ + eProsima_user_DllExport SummaryUpdate& operator =( + const SummaryUpdate& x) + { + + m_License = x.m_License; + + m_StableValue = x.m_StableValue; + + return *this; + } + + /*! + * @brief Move assignment. + * @param x Reference to the object SummaryUpdate that will be copied. + */ + eProsima_user_DllExport SummaryUpdate& operator =( + SummaryUpdate&& x) noexcept + { + + m_License = std::move(x.m_License); + m_StableValue = x.m_StableValue; + return *this; + } + + /*! + * @brief Comparison operator. + * @param x SummaryUpdate object to compare. + */ + eProsima_user_DllExport bool operator ==( + const SummaryUpdate& x) const + { + return (m_License == x.m_License && + m_StableValue == x.m_StableValue); + } + + /*! + * @brief Comparison operator. + * @param x SummaryUpdate object to compare. + */ + eProsima_user_DllExport bool operator !=( + const SummaryUpdate& x) const + { + return !(*this == x); + } + + /*! + * @brief This function copies the value in member License + * @param _License New value to be copied in member License + */ + eProsima_user_DllExport void License( + const std::string& _License) + { + m_License = _License; + } + + /*! + * @brief This function moves the value in member License + * @param _License New value to be moved in member License + */ + eProsima_user_DllExport void License( + std::string&& _License) + { + m_License = std::move(_License); + } + + /*! + * @brief This function returns a constant reference to member License + * @return Constant reference to member License + */ + eProsima_user_DllExport const std::string& License() const + { + return m_License; + } + + /*! + * @brief This function returns a reference to member License + * @return Reference to member License + */ + eProsima_user_DllExport std::string& License() + { + return m_License; + } + + + /*! + * @brief This function sets a value in member StableValue + * @param _StableValue New value for member StableValue + */ + eProsima_user_DllExport void StableValue( + float _StableValue) + { + m_StableValue = _StableValue; + } + + /*! + * @brief This function returns the value of member StableValue + * @return Value of member StableValue + */ + eProsima_user_DllExport float StableValue() const + { + return m_StableValue; + } + + /*! + * @brief This function returns a reference to member StableValue + * @return Reference to member StableValue + */ + eProsima_user_DllExport float& StableValue() + { + return m_StableValue; + } + + + +private: + + std::string m_License; + float m_StableValue{0.0}; + +}; /*! * @brief This class represents the structure ScaleInfo defined by the user in the IDL file. * @ingroup System diff --git a/lib/System.idl b/lib/System.idl index 998379a..4e0639b 100644 --- a/lib/System.idl +++ b/lib/System.idl @@ -2,6 +2,16 @@ module WeighingSystem { + // ============================================ + // Topic: onSummaryUpdate + // 汇总信息更新主题 + // ============================================ + struct SummaryUpdate { + string License; // 车牌号 + float StableValue; // 稳定重量 + }; + + // ============================================ // Topic: onScaleInfoUpdate // 称重设备信息更新主题 diff --git a/lib/SystemCdrAux.hpp b/lib/SystemCdrAux.hpp index cb7cae9..5785b3e 100644 --- a/lib/SystemCdrAux.hpp +++ b/lib/SystemCdrAux.hpp @@ -24,6 +24,9 @@ #include "System.hpp" +constexpr uint32_t WeighingSystem_SummaryUpdate_max_cdr_typesize {268UL}; +constexpr uint32_t WeighingSystem_SummaryUpdate_max_key_cdr_typesize {0UL}; + constexpr uint32_t WeighingSystem_LightsUpdate_max_cdr_typesize {12UL}; constexpr uint32_t WeighingSystem_LightsUpdate_max_key_cdr_typesize {0UL}; @@ -79,6 +82,10 @@ namespace fastcdr { class Cdr; class CdrSizeCalculator; +eProsima_user_DllExport void serialize_key( + eprosima::fastcdr::Cdr& scdr, + const WeighingSystem::SummaryUpdate& data); + eProsima_user_DllExport void serialize_key( eprosima::fastcdr::Cdr& scdr, const WeighingSystem::ScaleInfo& data); diff --git a/lib/SystemCdrAux.ipp b/lib/SystemCdrAux.ipp index 8932beb..8607287 100644 --- a/lib/SystemCdrAux.ipp +++ b/lib/SystemCdrAux.ipp @@ -34,6 +34,102 @@ using namespace eprosima::fastcdr::exception; namespace eprosima { namespace fastcdr { +template<> +eProsima_user_DllExport size_t calculate_serialized_size( + eprosima::fastcdr::CdrSizeCalculator& calculator, + const WeighingSystem::SummaryUpdate& data, + size_t& current_alignment) +{ + using namespace WeighingSystem; + + static_cast(data); + + eprosima::fastcdr::EncodingAlgorithmFlag previous_encoding = calculator.get_encoding(); + size_t calculated_size {calculator.begin_calculate_type_serialized_size( + eprosima::fastcdr::CdrVersion::XCDRv2 == calculator.get_cdr_version() ? + eprosima::fastcdr::EncodingAlgorithmFlag::DELIMIT_CDR2 : + eprosima::fastcdr::EncodingAlgorithmFlag::PLAIN_CDR, + current_alignment)}; + + + calculated_size += calculator.calculate_member_serialized_size(eprosima::fastcdr::MemberId(0), + data.License(), current_alignment); + + calculated_size += calculator.calculate_member_serialized_size(eprosima::fastcdr::MemberId(1), + data.StableValue(), current_alignment); + + + calculated_size += calculator.end_calculate_type_serialized_size(previous_encoding, current_alignment); + + return calculated_size; +} + +template<> +eProsima_user_DllExport void serialize( + eprosima::fastcdr::Cdr& scdr, + const WeighingSystem::SummaryUpdate& data) +{ + using namespace WeighingSystem; + + eprosima::fastcdr::Cdr::state current_state(scdr); + scdr.begin_serialize_type(current_state, + eprosima::fastcdr::CdrVersion::XCDRv2 == scdr.get_cdr_version() ? + eprosima::fastcdr::EncodingAlgorithmFlag::DELIMIT_CDR2 : + eprosima::fastcdr::EncodingAlgorithmFlag::PLAIN_CDR); + + scdr + << eprosima::fastcdr::MemberId(0) << data.License() + << eprosima::fastcdr::MemberId(1) << data.StableValue() +; + scdr.end_serialize_type(current_state); +} + +template<> +eProsima_user_DllExport void deserialize( + eprosima::fastcdr::Cdr& cdr, + WeighingSystem::SummaryUpdate& data) +{ + using namespace WeighingSystem; + + cdr.deserialize_type(eprosima::fastcdr::CdrVersion::XCDRv2 == cdr.get_cdr_version() ? + eprosima::fastcdr::EncodingAlgorithmFlag::DELIMIT_CDR2 : + eprosima::fastcdr::EncodingAlgorithmFlag::PLAIN_CDR, + [&data](eprosima::fastcdr::Cdr& dcdr, const eprosima::fastcdr::MemberId& mid) -> bool + { + bool ret_value = true; + switch (mid.id) + { + case 0: + dcdr >> data.License(); + break; + + case 1: + dcdr >> data.StableValue(); + break; + + default: + ret_value = false; + break; + } + return ret_value; + }); +} + +void serialize_key( + eprosima::fastcdr::Cdr& scdr, + const WeighingSystem::SummaryUpdate& data) +{ + using namespace WeighingSystem; + + static_cast(scdr); + static_cast(data); + scdr << data.License(); + + scdr << data.StableValue(); + +} + + template<> eProsima_user_DllExport size_t calculate_serialized_size( eprosima::fastcdr::CdrSizeCalculator& calculator, diff --git a/lib/SystemPubSubTypes.cxx b/lib/SystemPubSubTypes.cxx index ab44c09..6033a0c 100644 --- a/lib/SystemPubSubTypes.cxx +++ b/lib/SystemPubSubTypes.cxx @@ -32,6 +32,188 @@ using InstanceHandle_t = eprosima::fastdds::rtps::InstanceHandle_t; using DataRepresentationId_t = eprosima::fastdds::dds::DataRepresentationId_t; namespace WeighingSystem { + SummaryUpdatePubSubType::SummaryUpdatePubSubType() + { + set_name("WeighingSystem::SummaryUpdate"); + uint32_t type_size = WeighingSystem_SummaryUpdate_max_cdr_typesize; + type_size += static_cast(eprosima::fastcdr::Cdr::alignment(type_size, 4)); /* possible submessage alignment */ + max_serialized_type_size = type_size + 4; /*encapsulation*/ + is_compute_key_provided = false; + uint32_t key_length = WeighingSystem_SummaryUpdate_max_key_cdr_typesize > 16 ? WeighingSystem_SummaryUpdate_max_key_cdr_typesize : 16; + key_buffer_ = reinterpret_cast(malloc(key_length)); + memset(key_buffer_, 0, key_length); + } + + SummaryUpdatePubSubType::~SummaryUpdatePubSubType() + { + if (key_buffer_ != nullptr) + { + free(key_buffer_); + } + } + + bool SummaryUpdatePubSubType::serialize( + const void* const data, + SerializedPayload_t& payload, + DataRepresentationId_t data_representation) + { + const SummaryUpdate* p_type = static_cast(data); + + // Object that manages the raw buffer. + eprosima::fastcdr::FastBuffer fastbuffer(reinterpret_cast(payload.data), payload.max_size); + // Object that serializes the data. + eprosima::fastcdr::Cdr ser(fastbuffer, eprosima::fastcdr::Cdr::DEFAULT_ENDIAN, + data_representation == DataRepresentationId_t::XCDR_DATA_REPRESENTATION ? + eprosima::fastcdr::CdrVersion::XCDRv1 : eprosima::fastcdr::CdrVersion::XCDRv2); + payload.encapsulation = ser.endianness() == eprosima::fastcdr::Cdr::BIG_ENDIANNESS ? CDR_BE : CDR_LE; + ser.set_encoding_flag( + data_representation == DataRepresentationId_t::XCDR_DATA_REPRESENTATION ? + eprosima::fastcdr::EncodingAlgorithmFlag::PLAIN_CDR : + eprosima::fastcdr::EncodingAlgorithmFlag::DELIMIT_CDR2); + + try + { + // Serialize encapsulation + ser.serialize_encapsulation(); + // Serialize the object. + ser << *p_type; + ser.set_dds_cdr_options({0,0}); + } + catch (eprosima::fastcdr::exception::Exception& /*exception*/) + { + return false; + } + + // Get the serialized length + payload.length = static_cast(ser.get_serialized_data_length()); + return true; + } + + bool SummaryUpdatePubSubType::deserialize( + SerializedPayload_t& payload, + void* data) + { + try + { + // Convert DATA to pointer of your type + SummaryUpdate* p_type = static_cast(data); + + // Object that manages the raw buffer. + eprosima::fastcdr::FastBuffer fastbuffer(reinterpret_cast(payload.data), payload.length); + + // Object that deserializes the data. + eprosima::fastcdr::Cdr deser(fastbuffer, eprosima::fastcdr::Cdr::DEFAULT_ENDIAN); + + // Deserialize encapsulation. + deser.read_encapsulation(); + payload.encapsulation = deser.endianness() == eprosima::fastcdr::Cdr::BIG_ENDIANNESS ? CDR_BE : CDR_LE; + + // Deserialize the object. + deser >> *p_type; + } + catch (eprosima::fastcdr::exception::Exception& /*exception*/) + { + return false; + } + + return true; + } + + uint32_t SummaryUpdatePubSubType::calculate_serialized_size( + const void* const data, + DataRepresentationId_t data_representation) + { + try + { + eprosima::fastcdr::CdrSizeCalculator calculator( + data_representation == DataRepresentationId_t::XCDR_DATA_REPRESENTATION ? + eprosima::fastcdr::CdrVersion::XCDRv1 :eprosima::fastcdr::CdrVersion::XCDRv2); + size_t current_alignment {0}; + return static_cast(calculator.calculate_serialized_size( + *static_cast(data), current_alignment)) + + 4u /*encapsulation*/; + } + catch (eprosima::fastcdr::exception::Exception& /*exception*/) + { + return 0; + } + } + + void* SummaryUpdatePubSubType::create_data() + { + return reinterpret_cast(new SummaryUpdate()); + } + + void SummaryUpdatePubSubType::delete_data( + void* data) + { + delete(reinterpret_cast(data)); + } + + bool SummaryUpdatePubSubType::compute_key( + SerializedPayload_t& payload, + InstanceHandle_t& handle, + bool force_md5) + { + if (!is_compute_key_provided) + { + return false; + } + + SummaryUpdate data; + if (deserialize(payload, static_cast(&data))) + { + return compute_key(static_cast(&data), handle, force_md5); + } + + return false; + } + + bool SummaryUpdatePubSubType::compute_key( + const void* const data, + InstanceHandle_t& handle, + bool force_md5) + { + if (!is_compute_key_provided) + { + return false; + } + + const SummaryUpdate* p_type = static_cast(data); + + // Object that manages the raw buffer. + eprosima::fastcdr::FastBuffer fastbuffer(reinterpret_cast(key_buffer_), + WeighingSystem_SummaryUpdate_max_key_cdr_typesize); + + // Object that serializes the data. + eprosima::fastcdr::Cdr ser(fastbuffer, eprosima::fastcdr::Cdr::BIG_ENDIANNESS, eprosima::fastcdr::CdrVersion::XCDRv2); + ser.set_encoding_flag(eprosima::fastcdr::EncodingAlgorithmFlag::PLAIN_CDR2); + eprosima::fastcdr::serialize_key(ser, *p_type); + if (force_md5 || WeighingSystem_SummaryUpdate_max_key_cdr_typesize > 16) + { + md5_.init(); + md5_.update(key_buffer_, static_cast(ser.get_serialized_data_length())); + md5_.finalize(); + for (uint8_t i = 0; i < 16; ++i) + { + handle.value[i] = md5_.digest[i]; + } + } + else + { + for (uint8_t i = 0; i < 16; ++i) + { + handle.value[i] = key_buffer_[i]; + } + } + return true; + } + + void SummaryUpdatePubSubType::register_type_object_representation() + { + register_SummaryUpdate_type_identifier(type_identifiers_); + } + ScaleInfoPubSubType::ScaleInfoPubSubType() { set_name("WeighingSystem::ScaleInfo"); diff --git a/lib/SystemPubSubTypes.hpp b/lib/SystemPubSubTypes.hpp index 093759a..bbff5a8 100644 --- a/lib/SystemPubSubTypes.hpp +++ b/lib/SystemPubSubTypes.hpp @@ -40,6 +40,87 @@ namespace WeighingSystem { + /*! + * @brief This class represents the TopicDataType of the type SummaryUpdate defined by the user in the IDL file. + * @ingroup System + */ + class SummaryUpdatePubSubType : public eprosima::fastdds::dds::TopicDataType + { + public: + + typedef SummaryUpdate type; + + eProsima_user_DllExport SummaryUpdatePubSubType(); + + eProsima_user_DllExport ~SummaryUpdatePubSubType() override; + + eProsima_user_DllExport bool serialize( + const void* const data, + eprosima::fastdds::rtps::SerializedPayload_t& payload, + eprosima::fastdds::dds::DataRepresentationId_t data_representation) override; + + eProsima_user_DllExport bool deserialize( + eprosima::fastdds::rtps::SerializedPayload_t& payload, + void* data) override; + + eProsima_user_DllExport uint32_t calculate_serialized_size( + const void* const data, + eprosima::fastdds::dds::DataRepresentationId_t data_representation) override; + + eProsima_user_DllExport bool compute_key( + eprosima::fastdds::rtps::SerializedPayload_t& payload, + eprosima::fastdds::rtps::InstanceHandle_t& ihandle, + bool force_md5 = false) override; + + eProsima_user_DllExport bool compute_key( + const void* const data, + eprosima::fastdds::rtps::InstanceHandle_t& ihandle, + bool force_md5 = false) override; + + eProsima_user_DllExport void* create_data() override; + + eProsima_user_DllExport void delete_data( + void* data) override; + + //Register TypeObject representation in Fast DDS TypeObjectRegistry + eProsima_user_DllExport void register_type_object_representation() override; + + #ifdef TOPIC_DATA_TYPE_API_HAS_IS_BOUNDED + eProsima_user_DllExport inline bool is_bounded() const override + { + return false; + } + + #endif // TOPIC_DATA_TYPE_API_HAS_IS_BOUNDED + + #ifdef TOPIC_DATA_TYPE_API_HAS_IS_PLAIN + + eProsima_user_DllExport inline bool is_plain( + eprosima::fastdds::dds::DataRepresentationId_t data_representation) const override + { + static_cast(data_representation); + return false; + } + + #endif // TOPIC_DATA_TYPE_API_HAS_IS_PLAIN + + #ifdef TOPIC_DATA_TYPE_API_HAS_CONSTRUCT_SAMPLE + eProsima_user_DllExport inline bool construct_sample( + void* memory) const override + { + static_cast(memory); + return false; + } + + #endif // TOPIC_DATA_TYPE_API_HAS_CONSTRUCT_SAMPLE + + private: + + eprosima::fastdds::MD5 md5_; + unsigned char* key_buffer_; + + }; + /*! * @brief This class represents the TopicDataType of the type ScaleInfo defined by the user in the IDL file. * @ingroup System diff --git a/lib/SystemTypeObjectSupport.cxx b/lib/SystemTypeObjectSupport.cxx index ab59ab6..cf75694 100644 --- a/lib/SystemTypeObjectSupport.cxx +++ b/lib/SystemTypeObjectSupport.cxx @@ -40,6 +40,103 @@ using namespace eprosima::fastdds::dds::xtypes; namespace WeighingSystem { // TypeIdentifier is returned by reference: dependent structures/unions are registered in this same method +void register_SummaryUpdate_type_identifier( + TypeIdentifierPair& type_ids_SummaryUpdate) +{ + + ReturnCode_t return_code_SummaryUpdate {eprosima::fastdds::dds::RETCODE_OK}; + return_code_SummaryUpdate = + eprosima::fastdds::dds::DomainParticipantFactory::get_instance()->type_object_registry().get_type_identifiers( + "WeighingSystem::SummaryUpdate", type_ids_SummaryUpdate); + if (eprosima::fastdds::dds::RETCODE_OK != return_code_SummaryUpdate) + { + StructTypeFlag struct_flags_SummaryUpdate = TypeObjectUtils::build_struct_type_flag(eprosima::fastdds::dds::xtypes::ExtensibilityKind::APPENDABLE, + false, false); + QualifiedTypeName type_name_SummaryUpdate = "WeighingSystem::SummaryUpdate"; + eprosima::fastcdr::optional type_ann_builtin_SummaryUpdate; + eprosima::fastcdr::optional ann_custom_SummaryUpdate; + CompleteTypeDetail detail_SummaryUpdate = TypeObjectUtils::build_complete_type_detail(type_ann_builtin_SummaryUpdate, ann_custom_SummaryUpdate, type_name_SummaryUpdate.to_string()); + CompleteStructHeader header_SummaryUpdate; + header_SummaryUpdate = TypeObjectUtils::build_complete_struct_header(TypeIdentifier(), detail_SummaryUpdate); + CompleteStructMemberSeq member_seq_SummaryUpdate; + { + TypeIdentifierPair type_ids_License; + ReturnCode_t return_code_License {eprosima::fastdds::dds::RETCODE_OK}; + return_code_License = + eprosima::fastdds::dds::DomainParticipantFactory::get_instance()->type_object_registry().get_type_identifiers( + "anonymous_string_unbounded", type_ids_License); + + if (eprosima::fastdds::dds::RETCODE_OK != return_code_License) + { + { + SBound bound = 0; + StringSTypeDefn string_sdefn = TypeObjectUtils::build_string_s_type_defn(bound); + if (eprosima::fastdds::dds::RETCODE_BAD_PARAMETER == + TypeObjectUtils::build_and_register_s_string_type_identifier(string_sdefn, + "anonymous_string_unbounded", type_ids_License)) + { + EPROSIMA_LOG_ERROR(XTYPES_TYPE_REPRESENTATION, + "anonymous_string_unbounded already registered in TypeObjectRegistry for a different type."); + } + } + } + StructMemberFlag member_flags_License = TypeObjectUtils::build_struct_member_flag(eprosima::fastdds::dds::xtypes::TryConstructFailAction::DISCARD, + false, false, false, false); + MemberId member_id_License = 0x00000000; + bool common_License_ec {false}; + CommonStructMember common_License {TypeObjectUtils::build_common_struct_member(member_id_License, member_flags_License, TypeObjectUtils::retrieve_complete_type_identifier(type_ids_License, common_License_ec))}; + if (!common_License_ec) + { + EPROSIMA_LOG_ERROR(XTYPES_TYPE_REPRESENTATION, "Structure License member TypeIdentifier inconsistent."); + return; + } + MemberName name_License = "License"; + eprosima::fastcdr::optional member_ann_builtin_License; + ann_custom_SummaryUpdate.reset(); + CompleteMemberDetail detail_License = TypeObjectUtils::build_complete_member_detail(name_License, member_ann_builtin_License, ann_custom_SummaryUpdate); + CompleteStructMember member_License = TypeObjectUtils::build_complete_struct_member(common_License, detail_License); + TypeObjectUtils::add_complete_struct_member(member_seq_SummaryUpdate, member_License); + } + { + TypeIdentifierPair type_ids_StableValue; + ReturnCode_t return_code_StableValue {eprosima::fastdds::dds::RETCODE_OK}; + return_code_StableValue = + eprosima::fastdds::dds::DomainParticipantFactory::get_instance()->type_object_registry().get_type_identifiers( + "_float", type_ids_StableValue); + + if (eprosima::fastdds::dds::RETCODE_OK != return_code_StableValue) + { + EPROSIMA_LOG_ERROR(XTYPES_TYPE_REPRESENTATION, + "StableValue Structure member TypeIdentifier unknown to TypeObjectRegistry."); + return; + } + StructMemberFlag member_flags_StableValue = TypeObjectUtils::build_struct_member_flag(eprosima::fastdds::dds::xtypes::TryConstructFailAction::DISCARD, + false, false, false, false); + MemberId member_id_StableValue = 0x00000001; + bool common_StableValue_ec {false}; + CommonStructMember common_StableValue {TypeObjectUtils::build_common_struct_member(member_id_StableValue, member_flags_StableValue, TypeObjectUtils::retrieve_complete_type_identifier(type_ids_StableValue, common_StableValue_ec))}; + if (!common_StableValue_ec) + { + EPROSIMA_LOG_ERROR(XTYPES_TYPE_REPRESENTATION, "Structure StableValue member TypeIdentifier inconsistent."); + return; + } + MemberName name_StableValue = "StableValue"; + eprosima::fastcdr::optional member_ann_builtin_StableValue; + ann_custom_SummaryUpdate.reset(); + CompleteMemberDetail detail_StableValue = TypeObjectUtils::build_complete_member_detail(name_StableValue, member_ann_builtin_StableValue, ann_custom_SummaryUpdate); + CompleteStructMember member_StableValue = TypeObjectUtils::build_complete_struct_member(common_StableValue, detail_StableValue); + TypeObjectUtils::add_complete_struct_member(member_seq_SummaryUpdate, member_StableValue); + } + CompleteStructType struct_type_SummaryUpdate = TypeObjectUtils::build_complete_struct_type(struct_flags_SummaryUpdate, header_SummaryUpdate, member_seq_SummaryUpdate); + if (eprosima::fastdds::dds::RETCODE_BAD_PARAMETER == + TypeObjectUtils::build_and_register_struct_type_object(struct_type_SummaryUpdate, type_name_SummaryUpdate.to_string(), type_ids_SummaryUpdate)) + { + EPROSIMA_LOG_ERROR(XTYPES_TYPE_REPRESENTATION, + "WeighingSystem::SummaryUpdate already registered in TypeObjectRegistry for a different type."); + } + } +} +// TypeIdentifier is returned by reference: dependent structures/unions are registered in this same method void register_ScaleInfo_type_identifier( TypeIdentifierPair& type_ids_ScaleInfo) { diff --git a/lib/SystemTypeObjectSupport.hpp b/lib/SystemTypeObjectSupport.hpp index aec3f6e..922446b 100644 --- a/lib/SystemTypeObjectSupport.hpp +++ b/lib/SystemTypeObjectSupport.hpp @@ -38,6 +38,19 @@ #ifndef DOXYGEN_SHOULD_SKIP_THIS_PUBLIC namespace WeighingSystem { +/** + * @brief Register SummaryUpdate related TypeIdentifier. + * Fully-descriptive TypeIdentifiers are directly registered. + * Hash TypeIdentifiers require to fill the TypeObject information and hash it, consequently, the TypeObject is + * indirectly registered as well. + * + * @param[out] TypeIdentifier of the registered type. + * The returned TypeIdentifier corresponds to the complete TypeIdentifier in case of hashed TypeIdentifiers. + * Invalid TypeIdentifier is returned in case of error. + */ +eProsima_user_DllExport void register_SummaryUpdate_type_identifier( + eprosima::fastdds::dds::xtypes::TypeIdentifierPair& type_ids); + /** * @brief Register ScaleInfo related TypeIdentifier. * Fully-descriptive TypeIdentifiers are directly registered. diff --git a/weigh/Publisher.cxx b/weigh/Publisher.cxx index ecbe03e..486952e 100644 --- a/weigh/Publisher.cxx +++ b/weigh/Publisher.cxx @@ -93,7 +93,7 @@ PublisherApp::PublisherApp( publisher_->get_default_datawriter_qos(writer_qos); writer_qos.reliability().kind = ReliabilityQosPolicyKind::RELIABLE_RELIABILITY_QOS; writer_qos.reliability().max_blocking_time = Duration_t(1, 0); - writer_qos.durability().kind = DurabilityQosPolicyKind::TRANSIENT_LOCAL_DURABILITY_QOS; + writer_qos.durability().kind = DurabilityQosPolicyKind::VOLATILE_DURABILITY_QOS; writer_qos.history().kind = HistoryQosPolicyKind::KEEP_LAST_HISTORY_QOS; writer_qos.history().depth = 1; writer_qos.resource_limits().max_samples = 200; @@ -250,14 +250,14 @@ void PublisherApp::run(std::shared_ptr handler) if (detector.isStable()) { info.WeightOK() = 1; - info.Value() = weight; - info.StableValue() = weight; + info.Value() = weight/1000; + info.StableValue() = weight/1000; std::cout << "STABLE WEIGHT: " << detector.getStableWeight() << " kg" << std::endl; } else { info.WeightOK() = 0; - info.Value() = weight; + info.Value() = weight/1000; info.StableValue() = 0; }