pc-system/mqtt/mqttclient.cpp

#include "mqttclient.h"
#include <QJsonDocument>
#include <QJsonObject>
#include "../logger.h"
#include <QMessageBox>

MqttClient *MqttClient::instance = nullptr;

MqttClient::MqttClient(QObject *parent)
    : QObject(parent), mqClient(new QMqttClient(this))
{
    // 连接信号和槽
    connect(mqClient, &QMqttClient::connected, this, &MqttClient::onConnected);
    // 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 error: %1").arg(error)); });

    connect(mqClient, &QMqttClient::connected, this, &MqttClient::connected);
}

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

MqttClient *MqttClient::createNewInstance()
{
    return new MqttClient();
}
void MqttClient::connectToMqttBroker()
{
    qDebug() << "MQTT 1511: ";

    mqClient->setHostname("114.55.233.194");
    mqClient->setPort(1883);

    // // 设置账号和密码
    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->setUsername(username);
    mqClient->setPassword(password);

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

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

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)
            qDebug() << "Could not publish message";
        else
            qDebug() << "Message published" << message;
    }
    else
    {
        qDebug() << "Not connected to MQTT server";
    }
}

void MqttClient::onMessageReceived(const QByteArray &message, const QMqttTopicName &topic)
{
    if (topic.name() == "hxgc/companycode/pro/P")
    {
        emit proMessageReceived(message, topic);
    }
    else
    {
        emit messageAndTopicReceived(message, topic);
    }
}

void MqttClient::onStateChanged(QMqttClient::ClientState state)
{
    switch (state)
    {
    case QMqttClient::ClientState::Disconnected:
        QMessageBox::information(nullptr, "提示", "MQTT连接已断开，请重新登录");
        qDebug() << "MQTT 客户端状态: 断开连接";
        Logger::getInstance().info("Start init Mqtt server.");
        break;
    case QMqttClient::ClientState::Connecting:
        qDebug() << "MQTT 客户端状态: 正在连接";
        break;
    case QMqttClient::ClientState::Connected:
        qDebug() << "MQTT 客户端状态: 已连接";
        break;
    default:
        qDebug() << "MQTT 客户端状态: 未知状态";
        break;
    }
}

void MqttClient::onError(QMqttClient::ClientError error)
{

    qDebug() << "MQTT 错误: " << error;
}