fwt_software/MQTT.cpp

#include "MQTT.h"

const int	QOS = 1;
const int	N_RETRY_ATTEMPTS = 5;


void MQTTCallback::on_failure(const mqtt::token& tok)
{
    std::cout << "Connection failed" << std::endl;
    if (++nretry_ > N_RETRY_ATTEMPTS)
        std::cout << "Connection attempt failed already a few times" << std::endl;
    reconnect();
}


// (Re)connection success
void MQTTCallback::connected(const std::string& cause) {
    std::cout << "\nConnection success" << std::endl;
    std::cout << "\nSubscribing to topics.." << std::endl;

    cli_.subscribe(tpoic_target_hdg, QOS, nullptr, subListener_);
    cli_.subscribe(topic_control_mode, QOS, nullptr, subListener_);
}

void MQTTCallback::message_arrived(mqtt::const_message_ptr msg) {
    std::cout << "Message arrived" << std::endl;
    std::cout << "\ttopic: '" << msg->get_topic() << "'" << std::endl;
    std::cout << "\tpayload: '" << msg->to_string() << "'\n" << std::endl;
    if (msg->get_topic() == "GGS/FWT/" + fwt_name + "/target_HDG") {
        try {
            int value = static_cast<int>(std::stoi(msg->to_string()));
            std::unique_lock<std::mutex> ul(mqtt_mut);
            sub_data.set_target_heading(msg->to_string());
        }
        catch (const std::invalid_argument& e) {
            std::cerr << "MQTT type convertion invalid argument: " << e.what() << std::endl;
        }
        catch (const std::out_of_range& e) {
            std::cerr << "MQTT type convertion out of range: " << e.what() << std::endl;
        }
    }
    else if (msg->get_topic() == "GGS/FWT/" + fwt_name + "/ControlCode") {
        try {
            std::unique_lock<std::mutex> ul(mqtt_mut);
            sub_data.set_control_code(static_cast<int>(std::stoi(msg->to_string())));
        }
        catch (const std::invalid_argument& e) {
            std::cerr << "MQTT type convertion invalid argument: " << e.what() << std::endl;
        }
        catch (const std::out_of_range& e) {
            std::cerr << "MQTT type convertion out of range: " << e.what() << std::endl;
        }
    }
}

//////////////////////////////////////////////////////////////////////////////////////////////////////////////7x

MQTTClient::MQTTClient(const std::string& serverURI, const std::string& clientId, std::string tower_name, std::string login_user, std::string login_pw)
    : client(serverURI, clientId), callback(client, connOpts, tower_name)
{
    connOpts.set_keep_alive_interval(20);
    connOpts.set_clean_session(true);
    connOpts.set_user_name(login_user);
    connOpts.set_password(login_pw);
    std::cout << "MQTT object created" << std::endl;
    client.set_callback(callback);
    std::cout << "MQTT callback set" << std::endl;
}

MQTTClient::~MQTTClient() {
    if (client.is_connected())
        disconnect();
}

void MQTTClient::connect_client() {
    std::cout << "connecting MQTT..." << std::endl;
    running = true;
    mqtt_thread = std::thread(&MQTTClient::run, this);

}

void MQTTClient::run() {
    const auto TIMEOUT = std::chrono::seconds(5);
    try {
        conToken = client.connect(connOpts, nullptr, callback);
        conToken->wait_for(TIMEOUT);
        connected = true;
    }
    catch (const mqtt::exception& exc) {
        std::cerr << "Error: " << exc.what() << std::endl;
        running = false;
        connected = false;
        return;
    }
    std::cout << "MQTT connected" << std::endl;

    while (running)
        ;
}

void MQTTClient::disconnect() {
    // Disconnect
    try {
        std::cout << "\nDisconnecting from the MQTT server..." << std::flush;
        client.disconnect()->wait();
        std::cout << "OK" << std::endl;
    }
    catch (const mqtt::exception& exc) {
        std::cerr << exc.what() << std::endl;
    }
    running = false;
    mqtt_thread.join();
}

void MQTTClient::publish(const std::string& topic, const std::string& payload) {
    const auto TIMEOUT = std::chrono::seconds(3);
    mqtt::delivery_token_ptr pubtok;
    mqtt::message_ptr msg = mqtt::make_message(topic, payload);
    msg->set_qos(1);
    msg->set_retained(true);
    try {
        pubtok = client.publish(msg);
        pubtok->wait_for(TIMEOUT);
    }
    catch (const mqtt::exception& exc) {
        std::cerr << "Error: " << exc.what() << std::endl;
    }
}

void MQTTClient::subscribe(const std::string& topic) {
    try {
        subToken = client.subscribe(topic, 1);
        subToken->wait();
    }
    catch (const mqtt::exception& exc) {
        std::cerr << "Error: " << exc.what() << std::endl;
    }
}

void MQTTClient::receiveMessages() {
    try {
        client.start_consuming();
    }
    catch (const mqtt::exception& exc) {
        std::cerr << "Error: " << exc.what() << std::endl;
    }
    while (true) {

    }
}

bool MQTTClient::is_connected_to_server()
{
    return client.is_connected();
}