// 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" #include "DEBUG.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::mutex DdsMsgData::queue_cv_mtx_; std::queue MqttMsgData::Mqtt_msg_queue_; std::mutex MqttMsgData::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); DEBUG("Program is running. Please press Ctrl+C to stop at any time."); 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; }