pc-system/mqtt/mqttclient.cpp

#include "mqttclient.h"

#include <QJsonDocument>
#include <QJsonObject>
#include <QMessageBox>

#include "../global.h"
#include "../logger.h"

MqttClient *MqttClient::instance = nullptr;

MqttClient::MqttClient(QObject *parent) : QObject(parent), mqClient(new QMqttClient(this)) {
    // 连接信号和槽
    connect(mqClient, &QMqttClient::connected, this, &MqttClient::onConnected);
    // TODO: connect(mqClient, &QMqttClient::disconnected, this, &MqttClient::onDisconnected);
    connect(mqClient, &QMqttClient::stateChanged, this, &MqttClient::onStateChanged);
    connect(mqClient, &QMqttClient::errorChanged, this, &MqttClient::onError);
    connect(mqClient, &QMqttClient::messageReceived, this, &MqttClient::onMessageReceived);
    connect(mqClient, &QMqttClient::errorChanged, this, [this](QMqttClient::ClientError error) {
        Logger::getInstance().error(QString("MQTT client occured error: %1").arg(error));
    });
}

// 单例模式
MqttClient *MqttClient::getInstance() {
    if (instance == nullptr) {
        instance = new MqttClient();
    }
    return instance;
}

MqttClient *MqttClient::createNewInstance() { return new MqttClient(); }
void MqttClient::connectToMqttBroker() {
    mqClient->setHostname("114.55.233.194");
    mqClient->setPort(1883);
    mqClient->setKeepAlive(20);

    // // 设置账号和密码
    mqClient->setUsername("hxgc");
    mqClient->setPassword("hxgc123456");

    // 设置客户端 ID
    mqClient->setClientId("pc");
    mqClient->connectToHost();
}

void MqttClient::connectToMqttBroker(const QString &hostname, quint16 port, const QString &username,
                                     const QString &password, const QString &clientId,
                                     const QStringList &topicsToSubscribe) {
    mqClient->setHostname(hostname);
    mqClient->setPort(port);
    mqClient->setKeepAlive(20);

    // 设置账号和密码
    mqClient->setUsername(username);
    mqClient->setPassword(password);

    // 设置客户端 ID
    mqClient->setClientId(clientId);
    mqClient->connectToHost();
    // 保存要订阅的主题列表
    m_subscribeTopics = topicsToSubscribe;
}
void MqttClient::onConnected() {
    Logger::getInstance().debug("MQTT conncted");
    m_connectRetryCount = 0;
    subscribeToTopics(m_subscribeTopics);
}

void MqttClient::subscribeToTopics(const QStringList &topics) {
    for (const auto &topic : topics) {
        subscribeToTopic(topic);
    }
}

void MqttClient::disconnectFromMqttBroker() {
    Logger::getInstance().info("MQTT client disconnecting from broker" + mqClient->state());
    if (mqClient->state() == QMqttClient::Connected) {
        mqClient->disconnectFromHost();
    }
}

void MqttClient::subscribeToTopic(const QString &topic) { mqClient->subscribe(topic); }

void MqttClient::sendMessage(const QString &topic, const QByteArray &message, quint8 qos, bool isRetainedMsg) {
    if (mqClient->state() == QMqttClient::Connected) {
        auto pub = mqClient->publish(QMqttTopicName(topic), message, qos, isRetainedMsg);
        if (pub == -1)
            Logger::getInstance().error(
                QString("MQTT client sent message to topic: %1, msg: %2, qos: %3, isRetainedMsg: %4")
                    .arg(topic, QString(message), QString::number(qos), isRetainedMsg ? "true" : "false"));
        else
            Logger::getInstance().info(
                QString("MQTT client sent message to topic: %1, msg: %2, qos: %3, isRetainedMsg: %4")
                    .arg(topic, QString(message), QString::number(qos), isRetainedMsg ? "true" : "false"));
    } else {
        qDebug() << "Not connected to MQTT server";
        Logger::getInstance().error(
            QString("MQTT client is not connected to the server. topic: %1, msg: %2").arg(topic, QString(message)));
    }
}

void MqttClient::onMessageReceived(const QByteArray &message, const QMqttTopicName &topic) {
    if (topic.name() == MQTT_TOPIC_COMPANY_PROJECTS) {
        emit proMessageReceived(message, topic);
    } else {
        emit messageReceived(message, topic);
    }
}

void MqttClient::onStateChanged(QMqttClient::ClientState state) {
    switch (state) {
        case QMqttClient::ClientState::Disconnected:
            m_connectRetryCount++;
            if (m_connectRetryCount < 3) {
                Logger::getInstance().info(QString("Mqtt reconnecting... (%1)").arg(m_connectRetryCount));
                connectToMqttBroker();
            } else {
                Logger::getInstance().error("Mqtt reconnect failed after 3 attempts");
                QMessageBox::warning(nullptr, "提示", "mqtt连接已断开，请重新登录");
            }
            break;
        case QMqttClient::ClientState::Connecting:
            qDebug() << "MQTT 客户端状态: 正在连接";
            break;
        case QMqttClient::ClientState::Connected:
            m_connectRetryCount = 0;  // 重置连接重试计数
            qDebug() << "MQTT 客户端状态: 已连接";
            break;
        default:
            Logger::getInstance().info("Mqtt unkown state: " + QString::number(state));
            break;
    }
}

void MqttClient::onError(QMqttClient::ClientError error) {
    qDebug() << "MQTT 错误: " << error;
    Logger::getInstance().error(QString("MQTT client error: %1").arg(error));
}