youlai-boot/src/main/java/com/youlai/boot/config/MqttConfig.java

package com.youlai.boot.config;

import com.youlai.boot.config.property.MqttProperties;
import com.youlai.boot.support.mqtt.MqttMessageHandler;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.mqttv5.client.*;
import org.eclipse.paho.mqttv5.common.MqttException;
import org.eclipse.paho.mqttv5.common.MqttMessage;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import jakarta.annotation.PreDestroy;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;

/**
 * MQTT 配置
 * <p>
 * 基于 Eclipse Paho MQTT v5 原生客户端 (org.eclipse.paho.mqttv5.client) 实现消息的发布与订阅。
 * 通过 {@code mqtt.enabled=false} 可整体关闭 MQTT 功能。
 *
 * @author TongTongStudio
 * @since 2026/8/8
 */
@Slf4j
@Configuration
@ConditionalOnProperty(prefix = "mqtt", name = "enabled", havingValue = "true")
public class MqttConfig {

    /**
     * MQTT 客户端实例（保存引用，供 @PreDestroy 优雅关闭）
     */
    private IMqttClient client;

    /**
     * MQTT 客户端
     * <p>同一客户端同时用于发布与订阅。连接成功后立即订阅配置的主题。</p>
     */
    @Bean(destroyMethod = "disconnect")
    public IMqttClient mqttClient(MqttProperties properties, MqttMessageHandler mqttMessageHandler) throws MqttException {
        String serverUri = normalizeServerUri(properties.getUrl());
        String clientId = properties.getClientId();

        client = new MqttClient(serverUri, clientId);

        MqttConnectionOptionsBuilder builder = new MqttConnectionOptionsBuilder()
                .serverURI(serverUri)
                .connectionTimeout(properties.getConnectionTimeout())
                .keepAliveInterval(properties.getKeepAliveInterval())
                .automaticReconnect(properties.isAutomaticReconnect())
                .sessionExpiryInterval(TimeUnit.DAYS.toSeconds(1))
                .cleanStart(false);

        if (properties.getUsername() != null && !properties.getUsername().isEmpty()) {
            builder.username(properties.getUsername());
        }
        if (properties.getPassword() != null && !properties.getPassword().isEmpty()) {
            builder.password(properties.getPassword().getBytes(StandardCharsets.UTF_8));
        }

        MqttConnectionOptions options = builder.build();

        // 消息到达回调：分发到业务处理器
        client.setCallback(new org.eclipse.paho.mqttv5.client.MqttCallback() {
            @Override
            public void messageArrived(String topic, MqttMessage message) {
                String payload = new String(message.getPayload(), StandardCharsets.UTF_8);
                mqttMessageHandler.handleMessage(topic, payload);
            }

            @Override
            public void deliveryComplete(IMqttToken iMqttToken) {

            }

            @Override
            public void disconnected(org.eclipse.paho.mqttv5.client.MqttDisconnectResponse disconnectResponse) {
                log.warn("MQTT 连接断开: reasonCode={}, reasonString={}",
                        disconnectResponse.getReturnCode(), disconnectResponse.getReasonString());
            }

            @Override
            public void mqttErrorOccurred(MqttException exception) {
                log.error("MQTT 发生异常", exception);
            }

            @Override
            public void connectComplete(boolean reconnect, String serverURI) {
                log.info("MQTT 连接成功, reconnect={}, serverURI={}", reconnect, serverURI);
                // 连接/重连成功后订阅主题
                subscribeTopics(client, properties);
            }

            @Override
            public void authPacketArrived(int i, org.eclipse.paho.mqttv5.common.packet.MqttProperties mqttProperties) {

            }
        });

        client.connect(options);
        log.info("MQTT 客户端已启动, serverURI={}, clientId={}", serverUri, clientId);

        // 首次连接成功后订阅主题（重连场景由 connectComplete 处理）
        if (client.isConnected()) {
            subscribeTopics(client, properties);
        }

        return client;
    }

    /**
     * 应用关闭时优雅停止 MQTT 客户端。
     * <p>先 {@code disconnect} 断开连接，再 {@code close} 释放 Paho 后台线程池，
     * 避免容器已销毁（Redisson 已 shutdown）后回调线程仍在处理消息而报错。</p>
     */
    @PreDestroy
    public void destroy() {
        try {
            if (client != null) {
                if (client.isConnected()) {
                    client.disconnect();
                }
                client.close();
            }
        } catch (MqttException e) {
            log.warn("关闭 MQTT 客户端失败", e);
        }
    }

    /**
     * 订阅配置中的所有主题
     */
    private void subscribeTopics(IMqttClient client, MqttProperties properties) {
        for (String topic : properties.getTopics()) {
            try {
                client.subscribe(topic, properties.getQos());
                log.info("已订阅 MQTT 主题: {}", topic);
            } catch (MqttException e) {
                log.error("订阅 MQTT 主题失败: {}", topic, e);
            }
        }
    }

    /**
     * 将 Broker 地址统一规范为 Paho 支持的 scheme。
     * <p>Eclipse Paho 仅内置 {@code tcp://} 与 {@code ssl://} 两种 scheme，
     * 配置中常见的 {@code mqtt://} 需转换为 {@code tcp://}（明文）或 {@code ssl://}（加密）。</p>
     */
    private String normalizeServerUri(String uri) {
        if (uri == null || uri.isEmpty()) {
            return uri;
        }
        if (uri.startsWith("mqtt://")) {
            return "tcp://" + uri.substring("mqtt://".length());
        }
        if (uri.startsWith("mqtts://")) {
            return "ssl://" + uri.substring("mqtts://".length());
        }
        return uri;
    }
}