ts/TransmitterSwitch/common/qmqtt/mqttBase/MQTTBase.cpp

#include "MQTTBase.h"

#include <QDebug>


#include "qmqtt_message.h"
#include "spdlog/spdlog.h"


MQTTBase::MQTTBase(QObject* parent) : QObject(parent)
{
    /* 获取日志 */
    // m_logger = spdlog::get("MQTT");
    // if(m_logger == nullptr)
    // {
    //     SPDLOG_ERROR("获取MQTT logger 失败！");
    //     exit(-1);
    // }

    // m_client.setKeepAlive(2); // 设置心跳时间为5秒
    m_reconnectTimer.setSingleShot(false);

    /* 连接信号和槽 */
    connect(&m_client,SIGNAL(connected()),this,SLOT(do_connected()));
    connect(&m_client,SIGNAL(disconnected()),this,SLOT(do_disconnect()));
    connect(&m_client,SIGNAL(error(QMQTT::ClientError)),this,SLOT(do_error(QMQTT::ClientError)));
    connect(&m_client,SIGNAL(subscribed(QString,quint8)),this,SLOT(do_subscribed(QString, quint8)));

    connect(&m_client,SIGNAL(received(QMQTT::Message)),this,SLOT(do_received(QMQTT::Message)));
    connect(&m_client, &QMQTT::Client::received, this, &MQTTBase::signal_recvMessage);
    connect(&m_reconnectTimer, &QTimer::timeout, this, &MQTTBase::do_timeoutReconnect);
}

/* 设置地址 */
void MQTTBase::setIPAndPort(const QString& IP, int port)
{
    m_mqttIP = IP;
    m_mqttPort = port;
    QHostAddress addr = QHostAddress(IP);
    m_client.setHost(addr);
    m_client.setPort(port);
}

/* 设置订阅 */
void MQTTBase::addSubcribe(const QString& topic, int qos)
{
    if(qos > 2)
    {
        SPDLOG_ERROR("QoS值不合法:{}", qos);
        return;
    }
    if(m_client.connectionState() == QMQTT::ConnectionState::STATE_CONNECTED)
    {
        /* 已连接到MQTT，之前添加的主题已经订阅，现在添加的也直接添加订阅 */
        m_client.subscribe(topic, qos);
        m_mapTopic.insert(topic, qos);
    }else 
    {
        /* 还未连接到MQTT，直接加入到订阅列表 */
        m_mapTopic.insert(topic, qos);
    }
}

/* 设置自动重连 */
void MQTTBase::setAutoReconnect(bool isAuto)
{
    m_isAutoReconnect = isAuto;
    m_client.setAutoReconnect(isAuto);
}

/* 连接到服务器 */
void MQTTBase::connectToServer()
{
    if(m_client.connectionState() == QMQTT::ConnectionState::STATE_CONNECTED)
    {
        SPDLOG_INFO("MQTT已经连接到服务器，无需重复连接");
        return;
    }
    else if(m_client.connectionState() == QMQTT::ConnectionState::STATE_CONNECTING)
    {
        SPDLOG_INFO("MQTT正在连接到服务器，请稍后");
        return;
    }
    m_client.connectToHost();
}

/* 获取连接状态 */
QMQTT::ConnectionState MQTTBase::connectState()
{
    return m_client.connectionState();
}

/* 发送消息 */
bool MQTTBase::sendMessage(const QString& topic, const QByteArray& message, int qos)
{
    if(m_client.connectionState() != QMQTT::ConnectionState::STATE_CONNECTED)
    {
        SPDLOG_ERROR("MQTT未连接到服务器，发送消息失败");
        return false;
    }
    QMQTT::Message msg(0, topic, message, qos);
    auto ret = m_client.publish(msg);
    if(ret != 0)
    {
        SPDLOG_ERROR("发送消息失败:{}, 错误代码:{}", topic.toStdString(), ret);
        return false;
    }

    return true;
}

/* 发送消息，设置消息保留 */
bool MQTTBase::sendMessage(const QString& topic, const QByteArray& message, int qos, bool retain)
{
    if(m_client.connectionState() != QMQTT::ConnectionState::STATE_CONNECTED)
    {
        SPDLOG_ERROR("MQTT未连接到服务器，发送消息失败");
        return false;
    }
    QMQTT::Message msg;
    msg.setTopic(topic);
    msg.setPayload(message);
    msg.setQos(qos);
    msg.setRetain(retain); // 设置消息保留
    auto ret = m_client.publish(msg);
    if(ret != 0)
    {
        SPDLOG_ERROR("发送消息失败:{}, 错误代码:{}", topic.toStdString(), ret);
        return false;
    }

    return true;
}



/* 连接成功 */
void MQTTBase::do_connected()
{
    SPDLOG_INFO("MQTT IP:{} ,Port:{} 连接成功!", m_mqttIP.toStdString(), m_mqttPort);
    if(m_reconnectTimer.isActive())
    {
        m_reconnectTimer.stop();
    }
    /* 订阅所有的主题 */
    for(auto& it : m_mapTopic.keys())
    {
        auto qos = m_mapTopic.value(it);
        m_client.subscribe(it, qos);
    }
}

/* 断开连接 */
void MQTTBase::do_disconnect()
{
    SPDLOG_INFO("MQTT 断开连接!");
    if(m_isAutoReconnect)
    {
        SPDLOG_INFO("MQTT 自动重连已开启，正在尝试重新连接...");
        if(m_reconnectTimer.isActive())
        {
            m_reconnectTimer.stop();
        }
        m_reconnectTimer.start(6000);
        connectToServer();
    }
    emit signal_disconnected();
    // m_isConnected = false;
}

/* 错误 */
void MQTTBase::do_error(const QMQTT::ClientError error)
{
    SPDLOG_ERROR("MQTT 错误:{}", (int)error);
    // if(m_isAutoReconnect)
    // {
    //     SPDLOG_INFO("MQTT 自动重连已开启，正在尝试重新连接...");
    //     m_client.connectToHost();
    // }
}

/* 订阅成功 */
void MQTTBase::do_subscribed(const QString& topic, const quint8 qos)
{
    SPDLOG_INFO("MQTT 订阅:{}, QoS:{} 成功", topic.toStdString(), qos);
}

/* 接收到消息 */
void MQTTBase::do_received(const QMQTT::Message& message)
{
    // SPDLOG_INFO("MQTTBase接收到一条消息:{}", message.topic().toStdString());
    recvMessage(message);
}

/* 超时重连 */
void MQTTBase::do_timeoutReconnect()
{
    SPDLOG_INFO("MQTT 超时重连...");
    connectToServer();
}