MqttProxy.cpp

/**
 * @file MqttProxy.cpp
 * @brief An MQTT proxy to communicate on MQTT bus
 *
 * MQTT proxy implementation
 *
 * @author Amine BAGGA (2025)
 */
#include <iostream>
#include <chrono>
#include <mqtt/callback.h>
#include <thread>
#include "MqttProxy.h"
#include "Network.h"

using namespace std;

class action_listener : public virtual mqtt::iaction_listener
{
    void on_failure(const mqtt::token& tok) override {
        cout << "iaction_listener::on_failure";
        if (tok.get_message_id() != 0)
            cout << " for token: [" << tok.get_message_id() << "]";
        cout << endl;
    }

    void on_success(const mqtt::token& tok) override {
        cout << "iaction_listener::on_success";
        if (tok.get_message_id() != 0)
            cout << " for token: [" << tok.get_message_id() << "]";
        auto top = tok.get_topics();
        if (top && !top->empty())
            cout << "\ttoken topic: '" << (*top)[0] << "', ...";
        cout << endl;
    }

public:
    action_listener(){}
};

class Callback : public virtual mqtt::callback, public virtual mqtt::iaction_listener
{
    int m_nretry;
    mqtt::async_client& m_cli;
    mqtt::connect_options& m_connOpts;
    action_listener m_subListener;
    MqttProxy* m_mqttProxy;
    string m_topics;

    void reconnect() {
        this_thread::sleep_for(chrono::milliseconds(2500));
        try {
            m_cli.connect(m_connOpts, nullptr, *this);
        }
        catch (const mqtt::exception& exc) {
            cout << "Callback::reconnect Error : " << exc.what() << endl;
            exit(1);
        }
    }

    void on_failure(const mqtt::token& tok) override {
        cout << "Callback::on_failure Connection attempt failed" << endl;
        if (++m_nretry > 5)
            exit(1);
        reconnect();
    }

    void on_success(const mqtt::token& tok) override {}

    void connected(const string& cause) override {
        m_mqttProxy->setStartFlag(true);
        cout << "Callback::connected Success" << endl;
        m_cli.subscribe(m_topics, 1, nullptr, m_subListener);
    }

    void connection_lost(const string& cause) override {
        cout << "Callback::connection_lost" << endl;
        if (!cause.empty())
            cout << "\tcause: " << cause << endl;
        cout << "Reconnecting..." << endl;
        m_nretry = 0;
        reconnect();
    }

    void message_arrived(mqtt::const_message_ptr msg) override {
        cout << "Callback::message_arrived" << endl;
        cout << "\ttopic: '" << msg->get_topic() << "'" << endl;
        cout << "\tpayload: '" << msg->to_string() << "'\n" << endl;

        thread thd([this, msg]() {
            try {
                string topic = msg->get_topic();
                string payload = msg->to_string();
                this->m_mqttProxy->handleMqttMessage(topic, payload);
            } catch (const mqtt::exception& exc) {
                cout << "Callback::message_arrived subscribe error: " << exc.what() << endl;
            }
        });

        thd.detach();
    }

    void delivery_complete(mqtt::delivery_token_ptr token) override {}

public:
    Callback(mqtt::async_client& cli, mqtt::connect_options& connOpts, MqttProxy* proxy, const string& topics)
        : m_nretry(0), m_cli(cli), m_connOpts(connOpts), m_subListener(), m_mqttProxy(proxy), m_topics(topics) {
    }
};

MqttProxy::MqttProxy(): m_network(nullptr), m_client(BROKER_ADDRESS, CLIENT_ID) {
}

MqttProxy::~MqttProxy() {
}

void MqttProxy::setNetwork(Network* network) {
    m_network = network;
}

void MqttProxy::start(void) {
    mqtt::connect_options connOpts;
    connOpts.set_keep_alive_interval(20);
    connOpts.set_clean_session(true);
    connOpts.set_connect_timeout(30);

    Callback cb(m_client, connOpts, this, MqttProxy::NETWORK_TOPICS);
    m_client.set_callback(cb);
    
    try {
        m_client.connect(connOpts, nullptr, cb);
    } catch (const mqtt::exception& e) {
        cout << "MqttProxy::start Failed to connect to MQTT broker : " << e.what() << endl;
        return;
    }
    cout << "MqttProxy::start Success" << endl;
    while (true) {
        this_thread::sleep_for(chrono::milliseconds(100));
    }
}

void MqttProxy::setStartFlag(const bool start) {
    m_started = start;
    if (m_network == nullptr) {
        cout << "MqttProxy::setStartFlag Null network pointer" << endl;
        return;
    }
    m_network->getConfigs();
}

void MqttProxy::handleMqttMessage(const string& topic, const string& payload) {
    if (m_network == nullptr) {
        cout << "MqttProxy::handleMqttMessage Null network pointer" << endl;
        return;
    }
    m_network->setConfig(topic, payload);
}

void MqttProxy::publish(const string& topic, const string& payload) {
    if (!m_started) {
        cout << "MqttProxy::publish Cannot publish messages as not yet connected" << endl;
        return;
    }
    try {
        mqtt::message_ptr pubmsg = mqtt::make_message(topic, payload.c_str());
        pubmsg->set_qos(0);
        pubmsg->set_retained(true);
        m_client.publish(pubmsg)->wait_for(chrono::seconds(10));
    } catch (const mqtt::exception& e) {
        cout << "MqttProxy::publish Failed to publish on topic " << topic << " : " << e.what() << endl;
    }
}