Queue/mqttdds/main.cxx

// 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 <csignal>
#include <cstring>
#include <functional>
#include <iostream>
#include <stdexcept>
#include <thread>

#include <fastdds/dds/log/Log.hpp>

#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<void(int)> 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<std::string> topics,
        std::vector<int> 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<std::mutex> 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<std::string> sub_topic_;
    std::vector<int> sub_qos_;
};

std::queue<WeighingSystem::BarUpdate> DdsMsgData::BarUpdate_queue_;
std::queue<WeighingSystem::LightsUpdate> DdsMsgData::LightsUpdate_queue_;
std::queue<WeighingSystem::InfraredUpdate> DdsMsgData::InfraredUpdate_queue_;
std::queue<WeighingSystem::InfraredCommandUpdate> DdsMsgData::InfraredCommandUpdate_queue_;
std::queue<WeighingSystem::LicenseSnapUpdate> DdsMsgData::LicenseSnapUpdate_queue_;
std::queue<WeighingSystem::ScaleInfo> DdsMsgData::ScaleInfo_queue_;
std::mutex DdsMsgData::queue_cv_mtx_;
std::queue<mqtt::const_message_ptr> MqttMsgData::Mqtt_msg_queue_;
std::mutex MqttMsgData::queue_cv_mtx_;

std::string mqtt_topic_id = "0";

int main(int argc, char** argv)
{
    auto ret = EXIT_SUCCESS;
    std::shared_ptr<SubscriberApp> sub;
    std::shared_ptr<PublisherApp> pub;
    std::shared_ptr<mqtt::async_client> dev = nullptr;

    int domain_id = 0;
    std::string mqtt_server;
    std::string mqtt_username = "admin";
    std::string mqtt_password = "admin";
    std::string mqtt_id = "MqttDds_0";
    std::vector<std::string> topic;
    const std::vector<int> 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("MqttDds_") + std::string(argv[i]);
            mqtt_topic_id = 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<SubscriberApp>(domain_id);
    pub = std::make_shared<PublisherApp>(domain_id);

    std::shared_ptr<mqtt_callback> cb;
    if (mqtt_server != "")
    {
        dev = std::make_shared<mqtt::async_client>(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();

        topic.push_back("pound/" + mqtt_topic_id + "/#");
        cb = std::make_shared<mqtt_callback>(*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;
}