BJ_ZTEITS-Developers / parking-cloud-yudao

Commit 68437cd8308307c1b5a84475c64443dd0b69016a

Authored by haohao
1 parent 3f83569e

feat：【IoT 物联网】新增 MQTT WebSocket 协议支持

Inline Side-by-side
Showing 11 changed files with 2540 additions and 3 deletions
yudao-dependencies/pom.xml
  View file @68437cd
@@ -68,7 +68,7 @@ @@ -68,7 +68,7 @@
68 <netty.version>4.2.4.Final</netty.version> 68 <netty.version>4.2.4.Final</netty.version>
69 <mqtt.version>1.2.5</mqtt.version> 69 <mqtt.version>1.2.5</mqtt.version>
70 <pf4j-spring.version>0.9.0</pf4j-spring.version> 70 <pf4j-spring.version>0.9.0</pf4j-spring.version>
71 - <vertx.version>4.5.13</vertx.version> 71 + <vertx.version>4.5.21</vertx.version>
72 <!-- 三方云服务相关 --> 72 <!-- 三方云服务相关 -->
73 <awssdk.version>2.30.14</awssdk.version> 73 <awssdk.version>2.30.14</awssdk.version>
74 <justauth.version>1.16.7</justauth.version> 74 <justauth.version>1.16.7</justauth.version>
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayConfiguration.java
  View file @68437cd
@@ -10,6 +10,10 @@ import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttDownstreamSubscr @@ -10,6 +10,10 @@ import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttDownstreamSubscr
10 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttUpstreamProtocol; 10 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttUpstreamProtocol;
11 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.manager.IotMqttConnectionManager; 11 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.manager.IotMqttConnectionManager;
12 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router.IotMqttDownstreamHandler; 12 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router.IotMqttDownstreamHandler;
  13 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqttws.IotMqttWsDownstreamSubscriber;
  14 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqttws.IotMqttWsUpstreamProtocol;
  15 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqttws.manager.IotMqttWsConnectionManager;
  16 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqttws.router.IotMqttWsDownstreamHandler;
13 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.IotTcpDownstreamSubscriber; 17 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.IotTcpDownstreamSubscriber;
14 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.IotTcpUpstreamProtocol; 18 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.IotTcpUpstreamProtocol;
15 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.manager.IotTcpConnectionManager; 19 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.manager.IotTcpConnectionManager;
@@ -151,4 +155,42 @@ public class IotGatewayConfiguration { @@ -151,4 +155,42 @@ public class IotGatewayConfiguration {
151 155
152 } 156 }
153 157
  158 + /**
  159 + * IoT 网关 MQTT WebSocket 协议配置类
  160 + */
  161 + @Configuration
  162 + @ConditionalOnProperty(prefix = "yudao.iot.gateway.protocol.mqtt-ws", name = "enabled", havingValue = "true")
  163 + @Slf4j
  164 + public static class MqttWsProtocolConfiguration {
  165 +
  166 + @Bean(destroyMethod = "close")
  167 + public Vertx mqttWsVertx() {
  168 + return Vertx.vertx();
  169 + }
  170 +
  171 + @Bean
  172 + public IotMqttWsUpstreamProtocol iotMqttWsUpstreamProtocol(IotGatewayProperties gatewayProperties,
  173 + IotDeviceMessageService messageService,
  174 + IotMqttWsConnectionManager connectionManager,
  175 + Vertx mqttWsVertx) {
  176 + return new IotMqttWsUpstreamProtocol(gatewayProperties.getProtocol().getMqttWs(),
  177 + messageService, connectionManager, mqttWsVertx);
  178 + }
  179 +
  180 + @Bean
  181 + public IotMqttWsDownstreamHandler iotMqttWsDownstreamHandler(IotDeviceMessageService messageService,
  182 + IotDeviceService deviceService,
  183 + IotMqttWsConnectionManager connectionManager) {
  184 + return new IotMqttWsDownstreamHandler(messageService, deviceService, connectionManager);
  185 + }
  186 +
  187 + @Bean
  188 + public IotMqttWsDownstreamSubscriber iotMqttWsDownstreamSubscriber(IotMqttWsUpstreamProtocol mqttWsUpstreamProtocol,
  189 + IotMqttWsDownstreamHandler downstreamHandler,
  190 + IotMessageBus messageBus) {
  191 + return new IotMqttWsDownstreamSubscriber(mqttWsUpstreamProtocol, downstreamHandler, messageBus);
  192 + }
  193 +
  194 + }
  195 +
154 } 196 }
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayProperties.java
  View file @68437cd
@@ -88,6 +88,11 @@ public class IotGatewayProperties { @@ -88,6 +88,11 @@ public class IotGatewayProperties {
88 */ 88 */
89 private MqttProperties mqtt; 89 private MqttProperties mqtt;
90 90
  91 + /**
  92 + * MQTT WebSocket 组件配置
  93 + */
  94 + private MqttWsProperties mqttWs;
  95 +
91 } 96 }
92 97
93 @Data 98 @Data
@@ -402,4 +407,100 @@ public class IotGatewayProperties { @@ -402,4 +407,100 @@ public class IotGatewayProperties {
402 407
403 } 408 }
404 409
  410 + @Data
  411 + public static class MqttWsProperties {
  412 +
  413 + /**
  414 + * 是否开启
  415 + */
  416 + @NotNull(message = "是否开启不能为空")
  417 + private Boolean enabled;
  418 +
  419 + /**
  420 + * WebSocket 服务器端口（默认：8083）
  421 + */
  422 + private Integer port = 8083;
  423 +
  424 + /**
  425 + * WebSocket 路径（默认：/mqtt）
  426 + */
  427 + @NotEmpty(message = "WebSocket 路径不能为空")
  428 + private String path = "/mqtt";
  429 +
  430 + /**
  431 + * 最大消息大小（字节）
  432 + */
  433 + private Integer maxMessageSize = 8192;
  434 +
  435 + /**
  436 + * 连接超时时间（秒）
  437 + */
  438 + private Integer connectTimeoutSeconds = 60;
  439 +
  440 + /**
  441 + * 保持连接超时时间（秒）
  442 + */
  443 + private Integer keepAliveTimeoutSeconds = 300;
  444 +
  445 + /**
  446 + * 是否启用 SSL（wss://）
  447 + */
  448 + private Boolean sslEnabled = false;
  449 +
  450 + /**
  451 + * SSL 配置
  452 + */
  453 + private SslOptions sslOptions = new SslOptions();
  454 +
  455 + /**
  456 + * WebSocket 子协议（通常为 "mqtt" 或 "mqttv3.1"）
  457 + */
  458 + @NotEmpty(message = "WebSocket 子协议不能为空")
  459 + private String subProtocol = "mqtt";
  460 +
  461 + /**
  462 + * 最大帧大小（字节）
  463 + */
  464 + private Integer maxFrameSize = 65536;
  465 +
  466 + /**
  467 + * SSL 配置选项
  468 + */
  469 + @Data
  470 + public static class SslOptions {
  471 +
  472 + /**
  473 + * 密钥证书选项
  474 + */
  475 + private io.vertx.core.net.KeyCertOptions keyCertOptions;
  476 +
  477 + /**
  478 + * 信任选项
  479 + */
  480 + private io.vertx.core.net.TrustOptions trustOptions;
  481 +
  482 + /**
  483 + * SSL 证书路径
  484 + */
  485 + private String certPath;
  486 +
  487 + /**
  488 + * SSL 私钥路径
  489 + */
  490 + private String keyPath;
  491 +
  492 + /**
  493 + * 信任存储路径
  494 + */
  495 + private String trustStorePath;
  496 +
  497 + /**
  498 + * 信任存储密码
  499 + */
  500 + private String trustStorePassword;
  501 +
  502 + }
  503 +
  504 + }
  505 +
405 } 506 }
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/IotMqttWsDownstreamSubscriber.java 0 → 100644
  View file @68437cd
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.mqttws;
  2 +
  3 +import cn.hutool.core.util.StrUtil;
  4 +import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
  5 +import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber;
  6 +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
  7 +import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
  8 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqttws.router.IotMqttWsDownstreamHandler;
  9 +import jakarta.annotation.PostConstruct;
  10 +import lombok.extern.slf4j.Slf4j;
  11 +
  12 +/**
  13 + * IoT MQTT WebSocket 下行消息订阅器
  14 + * <p>
  15 + * 订阅消息总线的设备下行消息，并通过 WebSocket 发送到设备
  16 + *
  17 + * @author 芋道源码
  18 + */
  19 +@Slf4j
  20 +public class IotMqttWsDownstreamSubscriber implements IotMessageSubscriber<IotDeviceMessage> {
  21 +
  22 + private final IotMqttWsUpstreamProtocol upstreamProtocol;
  23 + private final IotMqttWsDownstreamHandler downstreamHandler;
  24 + private final IotMessageBus messageBus;
  25 +
  26 + public IotMqttWsDownstreamSubscriber(IotMqttWsUpstreamProtocol upstreamProtocol,
  27 + IotMqttWsDownstreamHandler downstreamHandler,
  28 + IotMessageBus messageBus) {
  29 + this.upstreamProtocol = upstreamProtocol;
  30 + this.downstreamHandler = downstreamHandler;
  31 + this.messageBus = messageBus;
  32 + }
  33 +
  34 + @PostConstruct
  35 + public void init() {
  36 + messageBus.register(this);
  37 + log.info("[init][MQTT WebSocket 下行消息订阅器已启动，topic: {}]", getTopic());
  38 + }
  39 +
  40 + @Override
  41 + public String getTopic() {
  42 + return IotDeviceMessageUtils.buildMessageBusGatewayDeviceMessageTopic(upstreamProtocol.getServerId());
  43 + }
  44 +
  45 + @Override
  46 + public String getGroup() {
  47 + // 保证点对点消费，需要保证独立的 Group，所以使用 Topic 作为 Group
  48 + return getTopic();
  49 + }
  50 +
  51 + @Override
  52 + public void onMessage(IotDeviceMessage message) {
  53 + log.debug("[onMessage][收到下行消息，deviceId: {}，method: {}]",
  54 + message.getDeviceId(), message.getMethod());
  55 + try {
  56 + // 1. 校验
  57 + String method = message.getMethod();
  58 + if (StrUtil.isBlank(method)) {
  59 + log.warn("[onMessage][消息方法为空，deviceId: {}]", message.getDeviceId());
  60 + return;
  61 + }
  62 +
  63 + // 2. 委托给下行处理器处理业务逻辑
  64 + boolean success = downstreamHandler.handleDownstreamMessage(message);
  65 + if (success) {
  66 + log.debug("[onMessage][下行消息处理成功，deviceId: {}，method: {}]",
  67 + message.getDeviceId(), message.getMethod());
  68 + } else {
  69 + log.warn("[onMessage][下行消息处理失败，deviceId: {}，method: {}]",
  70 + message.getDeviceId(), message.getMethod());
  71 + }
  72 + } catch (Exception e) {
  73 + log.error("[onMessage][处理下行消息失败，deviceId: {}，method: {}]",
  74 + message.getDeviceId(), message.getMethod(), e);
  75 + }
  76 + }
  77 +
  78 +}
  79 +
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/IotMqttWsUpstreamProtocol.java 0 → 100644
  View file @68437cd
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.mqttws;
  2 +
  3 +import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
  4 +import cn.iocoder.yudao.module.iot.gateway.config.IotGatewayProperties;
  5 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqttws.manager.IotMqttWsConnectionManager;
  6 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqttws.router.IotMqttWsUpstreamHandler;
  7 +import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
  8 +import io.vertx.core.Vertx;
  9 +import io.vertx.core.http.HttpServer;
  10 +import io.vertx.core.http.HttpServerOptions;
  11 +import io.vertx.core.http.ServerWebSocket;
  12 +import jakarta.annotation.PostConstruct;
  13 +import jakarta.annotation.PreDestroy;
  14 +import lombok.Getter;
  15 +import lombok.extern.slf4j.Slf4j;
  16 +
  17 +/**
  18 + * IoT 网关 MQTT WebSocket 协议：接收设备上行消息
  19 + * <p>
  20 + * 基于 Vert.x 实现 MQTT over WebSocket 服务端，支持：
  21 + * - 标准 MQTT 3.1.1 协议
  22 + * - WebSocket 协议升级
  23 + * - SSL/TLS 加密（wss://）
  24 + * - 设备认证与连接管理
  25 + * - QoS 0/1/2 消息质量保证
  26 + *
  27 + * @author 芋道源码
  28 + */
  29 +@Slf4j
  30 +public class IotMqttWsUpstreamProtocol {
  31 +
  32 + private final IotGatewayProperties.MqttWsProperties mqttWsProperties;
  33 +
  34 + private final IotDeviceMessageService messageService;
  35 +
  36 + private final IotMqttWsConnectionManager connectionManager;
  37 +
  38 + private final Vertx vertx;
  39 +
  40 + @Getter
  41 + private final String serverId;
  42 +
  43 + private HttpServer httpServer;
  44 +
  45 + public IotMqttWsUpstreamProtocol(IotGatewayProperties.MqttWsProperties mqttWsProperties,
  46 + IotDeviceMessageService messageService,
  47 + IotMqttWsConnectionManager connectionManager,
  48 + Vertx vertx) {
  49 + this.mqttWsProperties = mqttWsProperties;
  50 + this.messageService = messageService;
  51 + this.connectionManager = connectionManager;
  52 + this.vertx = vertx;
  53 + this.serverId = IotDeviceMessageUtils.generateServerId(mqttWsProperties.getPort());
  54 + }
  55 +
  56 + @PostConstruct
  57 + public void start() {
  58 + // 创建 HTTP 服务器选项
  59 + HttpServerOptions options = new HttpServerOptions()
  60 + .setPort(mqttWsProperties.getPort())
  61 + .setIdleTimeout(mqttWsProperties.getKeepAliveTimeoutSeconds())
  62 + .setMaxWebSocketFrameSize(mqttWsProperties.getMaxFrameSize())
  63 + .setMaxWebSocketMessageSize(mqttWsProperties.getMaxMessageSize())
  64 + // 配置 WebSocket 子协议支持
  65 + .addWebSocketSubProtocol(mqttWsProperties.getSubProtocol());
  66 +
  67 + // 配置 SSL（如果启用）
  68 + if (Boolean.TRUE.equals(mqttWsProperties.getSslEnabled())) {
  69 + options.setSsl(true)
  70 + .setKeyCertOptions(mqttWsProperties.getSslOptions().getKeyCertOptions())
  71 + .setTrustOptions(mqttWsProperties.getSslOptions().getTrustOptions());
  72 + log.info("[start][MQTT WebSocket 已启用 SSL/TLS (wss://)]");
  73 + }
  74 +
  75 + // 创建 HTTP 服务器
  76 + httpServer = vertx.createHttpServer(options);
  77 +
  78 + // 设置 WebSocket 处理器
  79 + httpServer.webSocketHandler(this::handleWebSocketConnection);
  80 +
  81 + // 启动服务器
  82 + try {
  83 + httpServer.listen().result();
  84 + log.info("[start][IoT 网关 MQTT WebSocket 协议启动成功，端口: {}，路径: {}，支持子协议: {}]",
  85 + mqttWsProperties.getPort(), mqttWsProperties.getPath(),
  86 + "mqtt, mqttv3.1, " + mqttWsProperties.getSubProtocol());
  87 + } catch (Exception e) {
  88 + log.error("[start][IoT 网关 MQTT WebSocket 协议启动失败]", e);
  89 + throw e;
  90 + }
  91 + }
  92 +
  93 + @PreDestroy
  94 + public void stop() {
  95 + if (httpServer != null) {
  96 + try {
  97 + // 关闭所有连接
  98 + connectionManager.closeAllConnections();
  99 +
  100 + // 关闭服务器
  101 + httpServer.close().result();
  102 + log.info("[stop][IoT 网关 MQTT WebSocket 协议已停止]");
  103 + } catch (Exception e) {
  104 + log.error("[stop][IoT 网关 MQTT WebSocket 协议停止失败]", e);
  105 + }
  106 + }
  107 + }
  108 +
  109 + /**
  110 + * 处理 WebSocket 连接请求
  111 + *
  112 + * @param socket WebSocket 连接
  113 + */
  114 + private void handleWebSocketConnection(ServerWebSocket socket) {
  115 + String path = socket.path();
  116 + String subProtocol = socket.subProtocol();
  117 +
  118 + log.info("[handleWebSocketConnection][收到 WebSocket 连接请求，path: {}，subProtocol: {}，remoteAddress: {}]",
  119 + path, subProtocol, socket.remoteAddress());
  120 +
  121 + // 验证路径
  122 + if (!mqttWsProperties.getPath().equals(path)) {
  123 + log.warn("[handleWebSocketConnection][WebSocket 路径不匹配，拒绝连接，path: {}，期望: {}]",
  124 + path, mqttWsProperties.getPath());
  125 + socket.close();
  126 + return;
  127 + }
  128 +
  129 + // 验证子协议
  130 + // Vert.x 已经自动进行了子协议协商，这里只需要验证是否为 MQTT 相关协议
  131 + if (subProtocol != null && !subProtocol.startsWith("mqtt")) {
  132 + log.warn("[handleWebSocketConnection][WebSocket 子协议不支持，拒绝连接，subProtocol: {}]", subProtocol);
  133 + socket.close();
  134 + return;
  135 + }
  136 +
  137 + log.info("[handleWebSocketConnection][WebSocket 连接已接受，remoteAddress: {}，subProtocol: {}]",
  138 + socket.remoteAddress(), subProtocol);
  139 +
  140 + // 创建处理器并处理连接
  141 + IotMqttWsUpstreamHandler handler = new IotMqttWsUpstreamHandler(
  142 + this, messageService, connectionManager);
  143 + handler.handle(socket);
  144 + }
  145 +
  146 +}
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/manager/IotMqttWsConnectionManager.java 0 → 100644
  View file @68437cd
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.mqttws.manager;
  2 +
  3 +import io.vertx.core.http.ServerWebSocket;
  4 +import lombok.extern.slf4j.Slf4j;
  5 +import org.springframework.stereotype.Component;
  6 +
  7 +import java.util.Map;
  8 +import java.util.Set;
  9 +import java.util.concurrent.ConcurrentHashMap;
  10 +import java.util.concurrent.CopyOnWriteArraySet;
  11 +
  12 +/**
  13 + * IoT MQTT WebSocket 连接管理器
  14 + *
  15 + * @author 芋道源码
  16 + */
  17 +@Slf4j
  18 +@Component
  19 +public class IotMqttWsConnectionManager {
  20 +
  21 + /**
  22 + * 存储设备连接
  23 + * Key: 设备标识（deviceKey）
  24 + * Value: WebSocket 连接
  25 + */
  26 + private final Map<String, ServerWebSocket> connections = new ConcurrentHashMap<>();
  27 +
  28 + /**
  29 + * 存储设备标识与 Socket ID 的映射
  30 + * Key: 设备标识（deviceKey）
  31 + * Value: Socket ID（UUID）
  32 + */
  33 + private final Map<String, String> deviceKeyToSocketId = new ConcurrentHashMap<>();
  34 +
  35 + /**
  36 + * 存储 Socket ID 与设备标识的映射
  37 + * Key: Socket ID（UUID）
  38 + * Value: 设备标识（deviceKey）
  39 + */
  40 + private final Map<String, String> socketIdToDeviceKey = new ConcurrentHashMap<>();
  41 +
  42 + /**
  43 + * 存储设备订阅的主题
  44 + * Key: 设备标识（deviceKey）
  45 + * Value: 订阅的主题集合
  46 + */
  47 + private final Map<String, Set<String>> deviceSubscriptions = new ConcurrentHashMap<>();
  48 +
  49 + /**
  50 + * 添加连接
  51 + *
  52 + * @param deviceKey 设备标识
  53 + * @param socket WebSocket 连接
  54 + * @param socketId Socket ID（UUID）
  55 + */
  56 + public void addConnection(String deviceKey, ServerWebSocket socket, String socketId) {
  57 + connections.put(deviceKey, socket);
  58 + deviceKeyToSocketId.put(deviceKey, socketId);
  59 + socketIdToDeviceKey.put(socketId, deviceKey);
  60 + log.info("[addConnection][设备连接已添加，deviceKey: {}，socketId: {}，当前连接数: {}]",
  61 + deviceKey, socketId, connections.size());
  62 + }
  63 +
  64 + /**
  65 + * 移除连接
  66 + *
  67 + * @param deviceKey 设备标识
  68 + */
  69 + public void removeConnection(String deviceKey) {
  70 + ServerWebSocket socket = connections.remove(deviceKey);
  71 + String socketId = deviceKeyToSocketId.remove(deviceKey);
  72 + if (socketId != null) {
  73 + socketIdToDeviceKey.remove(socketId);
  74 + }
  75 + if (socket != null) {
  76 + log.info("[removeConnection][设备连接已移除，deviceKey: {}，socketId: {}，当前连接数: {}]",
  77 + deviceKey, socketId, connections.size());
  78 + }
  79 + }
  80 +
  81 + /**
  82 + * 根据 Socket ID 移除连接
  83 + *
  84 + * @param socketId WebSocket 文本框架 ID
  85 + */
  86 + public void removeConnectionBySocketId(String socketId) {
  87 + String deviceKey = socketIdToDeviceKey.remove(socketId);
  88 + if (deviceKey != null) {
  89 + connections.remove(deviceKey);
  90 + log.info("[removeConnectionBySocketId][设备连接已移除，socketId: {}，deviceKey: {}，当前连接数: {}]",
  91 + socketId, deviceKey, connections.size());
  92 + }
  93 + }
  94 +
  95 + /**
  96 + * 获取连接
  97 + *
  98 + * @param deviceKey 设备标识
  99 + * @return WebSocket 连接
  100 + */
  101 + public ServerWebSocket getConnection(String deviceKey) {
  102 + return connections.get(deviceKey);
  103 + }
  104 +
  105 + /**
  106 + * 根据 Socket ID 获取设备标识
  107 + *
  108 + * @param socketId WebSocket 文本框架 ID
  109 + * @return 设备标识
  110 + */
  111 + public String getDeviceKeyBySocketId(String socketId) {
  112 + return socketIdToDeviceKey.get(socketId);
  113 + }
  114 +
  115 + /**
  116 + * 检查设备是否在线
  117 + *
  118 + * @param deviceKey 设备标识
  119 + * @return 是否在线
  120 + */
  121 + public boolean isOnline(String deviceKey) {
  122 + return connections.containsKey(deviceKey);
  123 + }
  124 +
  125 + /**
  126 + * 获取当前连接数
  127 + *
  128 + * @return 连接数
  129 + */
  130 + public int getConnectionCount() {
  131 + return connections.size();
  132 + }
  133 +
  134 + /**
  135 + * 关闭所有连接
  136 + */
  137 + public void closeAllConnections() {
  138 + connections.forEach((deviceKey, socket) -> {
  139 + try {
  140 + socket.close();
  141 + log.info("[closeAllConnections][关闭设备连接，deviceKey: {}]", deviceKey);
  142 + } catch (Exception e) {
  143 + log.error("[closeAllConnections][关闭设备连接失败，deviceKey: {}]", deviceKey, e);
  144 + }
  145 + });
  146 + connections.clear();
  147 + deviceKeyToSocketId.clear();
  148 + socketIdToDeviceKey.clear();
  149 + deviceSubscriptions.clear();
  150 + log.info("[closeAllConnections][所有连接已关闭]");
  151 + }
  152 +
  153 + // ==================== 订阅管理方法 ====================
  154 +
  155 + /**
  156 + * 添加订阅
  157 + *
  158 + * @param deviceKey 设备标识
  159 + * @param topic 订阅主题
  160 + */
  161 + public void addSubscription(String deviceKey, String topic) {
  162 + deviceSubscriptions.computeIfAbsent(deviceKey, k -> new CopyOnWriteArraySet<>()).add(topic);
  163 + log.debug("[addSubscription][设备订阅主题，deviceKey: {}，topic: {}]", deviceKey, topic);
  164 + }
  165 +
  166 + /**
  167 + * 移除订阅
  168 + *
  169 + * @param deviceKey 设备标识
  170 + * @param topic 订阅主题
  171 + */
  172 + public void removeSubscription(String deviceKey, String topic) {
  173 + Set<String> topics = deviceSubscriptions.get(deviceKey);
  174 + if (topics != null) {
  175 + topics.remove(topic);
  176 + log.debug("[removeSubscription][设备取消订阅，deviceKey: {}，topic: {}]", deviceKey, topic);
  177 + }
  178 + }
  179 +
  180 + /**
  181 + * 检查设备是否订阅了指定主题
  182 + * 支持 MQTT 通配符匹配（+ 和 #）
  183 + *
  184 + * @param deviceKey 设备标识
  185 + * @param topic 发布主题
  186 + * @return 是否匹配
  187 + */
  188 + public boolean isSubscribed(String deviceKey, String topic) {
  189 + Set<String> subscriptions = deviceSubscriptions.get(deviceKey);
  190 + if (subscriptions == null || subscriptions.isEmpty()) {
  191 + return false;
  192 + }
  193 +
  194 + // 检查是否有匹配的订阅
  195 + for (String subscription : subscriptions) {
  196 + if (topicMatches(subscription, topic)) {
  197 + return true;
  198 + }
  199 + }
  200 + return false;
  201 + }
  202 +
  203 + /**
  204 + * 获取设备的所有订阅
  205 + *
  206 + * @param deviceKey 设备标识
  207 + * @return 订阅主题集合
  208 + */
  209 + public Set<String> getSubscriptions(String deviceKey) {
  210 + return deviceSubscriptions.get(deviceKey);
  211 + }
  212 +
  213 + /**
  214 + * MQTT 主题匹配
  215 + * 支持通配符：
  216 + * - +：匹配单层主题
  217 + * - #：匹配多层主题（必须在末尾）
  218 + *
  219 + * @param subscription 订阅主题（可能包含通配符）
  220 + * @param topic 发布主题（不包含通配符）
  221 + * @return 是否匹配
  222 + */
  223 + private boolean topicMatches(String subscription, String topic) {
  224 + // 完全匹配
  225 + if (subscription.equals(topic)) {
  226 + return true;
  227 + }
  228 +
  229 + // 不包含通配符
  230 + if (!subscription.contains("+") && !subscription.contains("#")) {
  231 + return false;
  232 + }
  233 +
  234 + String[] subscriptionParts = subscription.split("/");
  235 + String[] topicParts = topic.split("/");
  236 +
  237 + int i = 0;
  238 + for (; i < subscriptionParts.length && i < topicParts.length; i++) {
  239 + String subPart = subscriptionParts[i];
  240 + String topicPart = topicParts[i];
  241 +
  242 + if (subPart.equals("#")) {
  243 + // # 匹配剩余所有层级，且必须在末尾
  244 + return i == subscriptionParts.length - 1;
  245 + }
  246 +
  247 + if (!subPart.equals("+") && !subPart.equals(topicPart)) {
  248 + // 不是通配符且不匹配
  249 + return false;
  250 + }
  251 + }
  252 +
  253 + // 检查是否都匹配完
  254 + return i == subscriptionParts.length && i == topicParts.length;
  255 + }
  256 +
  257 +}
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/package-info.java 0 → 100644
  View file @68437cd
  1 +/**
  2 + * IoT 网关 MQTT WebSocket 协议实现
  3 + * <p>
  4 + * 基于 Vert.x 实现 MQTT over WebSocket 服务端，支持：
  5 + * - 标准 MQTT 3.1.1 协议
  6 + * - WebSocket 协议升级
  7 + * - SSL/TLS 加密（wss://）
  8 + * - 设备认证与连接管理
  9 + * - QoS 0/1/2 消息质量保证
  10 + * - 双向消息通信（上行/下行）
  11 + *
  12 + * @author 芋道源码
  13 + */
  14 +package cn.iocoder.yudao.module.iot.gateway.protocol.mqttws;
  15 +
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/router/IotMqttWsDownstreamHandler.java 0 → 100644
  View file @68437cd
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.mqttws.router;
  2 +
  3 +import cn.hutool.core.util.StrUtil;
  4 +import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceRespDTO;
  5 +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
  6 +import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
  7 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqttws.manager.IotMqttWsConnectionManager;
  8 +import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceService;
  9 +import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
  10 +import cn.iocoder.yudao.module.iot.gateway.util.IotMqttTopicUtils;
  11 +import io.vertx.core.buffer.Buffer;
  12 +import io.vertx.core.http.ServerWebSocket;
  13 +import lombok.extern.slf4j.Slf4j;
  14 +
  15 +import java.util.concurrent.atomic.AtomicInteger;
  16 +
  17 +/**
  18 + * IoT MQTT WebSocket 下行消息处理器
  19 + * <p>
  20 + * 处理从消息总线发送到设备的消息，包括：
  21 + * - 属性设置
  22 + * - 服务调用
  23 + * - 事件通知
  24 + *
  25 + * @author 芋道源码
  26 + */
  27 +@Slf4j
  28 +public class IotMqttWsDownstreamHandler {
  29 +
  30 + private final IotDeviceMessageService deviceMessageService;
  31 +
  32 + private final IotDeviceService deviceService;
  33 +
  34 + private final IotMqttWsConnectionManager connectionManager;
  35 +
  36 + /**
  37 + * 消息 ID 生成器（用于发布消息）
  38 + */
  39 + private final AtomicInteger messageIdGenerator = new AtomicInteger(1);
  40 +
  41 + public IotMqttWsDownstreamHandler(IotDeviceMessageService deviceMessageService,
  42 + IotDeviceService deviceService,
  43 + IotMqttWsConnectionManager connectionManager) {
  44 + this.deviceMessageService = deviceMessageService;
  45 + this.deviceService = deviceService;
  46 + this.connectionManager = connectionManager;
  47 + }
  48 +
  49 + /**
  50 + * 处理下行消息
  51 + *
  52 + * @param message 设备消息
  53 + * @return 是否处理成功
  54 + */
  55 + public boolean handleDownstreamMessage(IotDeviceMessage message) {
  56 + try {
  57 + // 1. 基础校验
  58 + if (message == null || message.getDeviceId() == null) {
  59 + log.warn("[handleDownstreamMessage][消息或设备 ID 为空，忽略处理]");
  60 + return false;
  61 + }
  62 +
  63 + // 2. 获取设备信息
  64 + IotDeviceRespDTO deviceInfo = deviceService.getDeviceFromCache(message.getDeviceId());
  65 + if (deviceInfo == null) {
  66 + log.warn("[handleDownstreamMessage][设备不存在，设备 ID：{}]", message.getDeviceId());
  67 + return false;
  68 + }
  69 +
  70 + // 3. 构建设备标识
  71 + String deviceKey = deviceInfo.getProductKey() + ":" + deviceInfo.getDeviceName();
  72 +
  73 + // 4. 检查设备是否在线
  74 + if (!connectionManager.isOnline(deviceKey)) {
  75 + log.warn("[handleDownstreamMessage][设备离线，无法发送消息，deviceKey: {}]", deviceKey);
  76 + return false;
  77 + }
  78 +
  79 + // 5. 构建主题
  80 + String topic = buildDownstreamTopic(message, deviceInfo);
  81 + if (StrUtil.isBlank(topic)) {
  82 + log.warn("[handleDownstreamMessage][主题构建失败，设备 ID：{}，方法：{}]",
  83 + message.getDeviceId(), message.getMethod());
  84 + return false;
  85 + }
  86 +
  87 + // 6. 检查设备是否订阅了该主题
  88 + if (!connectionManager.isSubscribed(deviceKey, topic)) {
  89 + log.warn("[handleDownstreamMessage][设备未订阅该主题，deviceKey: {}，topic: {}]", deviceKey, topic);
  90 + return false;
  91 + }
  92 +
  93 + // 8. 编码消息
  94 + byte[] payload = deviceMessageService.encodeDeviceMessage(message,
  95 + deviceInfo.getProductKey(), deviceInfo.getDeviceName());
  96 + if (payload == null || payload.length == 0) {
  97 + log.warn("[handleDownstreamMessage][消息编码失败，设备 ID：{}]", message.getDeviceId());
  98 + return false;
  99 + }
  100 +
  101 + // 9. 发送消息到设备
  102 + return sendMessageToDevice(deviceKey, topic, payload, 1);
  103 + } catch (Exception e) {
  104 + if (message != null) {
  105 + log.error("[handleDownstreamMessage][处理下行消息异常，设备 ID：{}，错误：{}]",
  106 + message.getDeviceId(), e.getMessage(), e);
  107 + }
  108 + return false;
  109 + }
  110 + }
  111 +
  112 + /**
  113 + * 构建下行消息主题
  114 + *
  115 + * @param message 设备消息
  116 + * @param deviceInfo 设备信息
  117 + * @return 主题
  118 + */
  119 + private String buildDownstreamTopic(IotDeviceMessage message, IotDeviceRespDTO deviceInfo) {
  120 + String method = message.getMethod();
  121 + if (StrUtil.isBlank(method)) {
  122 + return null;
  123 + }
  124 +
  125 + // 使用工具类构建主题，支持回复消息处理
  126 + boolean isReply = IotDeviceMessageUtils.isReplyMessage(message);
  127 + return IotMqttTopicUtils.buildTopicByMethod(method, deviceInfo.getProductKey(),
  128 + deviceInfo.getDeviceName(), isReply);
  129 + }
  130 +
  131 + /**
  132 + * 发送消息到设备
  133 + *
  134 + * @param deviceKey 设备标识（productKey:deviceName）
  135 + * @param topic 主题
  136 + * @param payload 消息内容
  137 + * @param qos QoS 级别
  138 + * @return 是否发送成功
  139 + */
  140 + private boolean sendMessageToDevice(String deviceKey, String topic, byte[] payload, int qos) {
  141 + // 获取设备连接
  142 + ServerWebSocket socket = connectionManager.getConnection(deviceKey);
  143 + if (socket == null) {
  144 + log.warn("[sendMessageToDevice][设备未连接，deviceKey: {}]", deviceKey);
  145 + return false;
  146 + }
  147 +
  148 + try {
  149 + int messageId = qos > 0 ? generateMessageId() : 0;
  150 +
  151 + // 手动编码MQTT PUBLISH消息
  152 + io.netty.buffer.ByteBuf byteBuf = io.netty.buffer.Unpooled.buffer();
  153 +
  154 + // 固定头：消息类型(PUBLISH=3) + DUP(0) + QoS + RETAIN
  155 + int fixedHeaderByte1 = 0x30 | (qos << 1); // PUBLISH类型
  156 + byteBuf.writeByte(fixedHeaderByte1);
  157 +
  158 + // 计算剩余长度
  159 + int topicLength = topic.getBytes().length;
  160 + int remainingLength = 2 + topicLength + (qos > 0 ? 2 : 0) + payload.length;
  161 +
  162 + // 写入剩余长度（简化版本，假设小于128字节）
  163 + if (remainingLength < 128) {
  164 + byteBuf.writeByte(remainingLength);
  165 + } else {
  166 + // 处理大于127的情况
  167 + int x = remainingLength;
  168 + do {
  169 + int encodedByte = x % 128;
  170 + x = x / 128;
  171 + if (x > 0) {
  172 + encodedByte = encodedByte | 128;
  173 + }
  174 + byteBuf.writeByte(encodedByte);
  175 + } while (x > 0);
  176 + }
  177 +
  178 + // 可变头：主题名称
  179 + byteBuf.writeShort(topicLength);
  180 + byteBuf.writeBytes(topic.getBytes());
  181 +
  182 + // 可变头：消息ID（仅QoS>0时）
  183 + if (qos > 0) {
  184 + byteBuf.writeShort(messageId);
  185 + }
  186 +
  187 + // 有效载荷
  188 + byteBuf.writeBytes(payload);
  189 +
  190 + // 发送
  191 + byte[] bytes = new byte[byteBuf.readableBytes()];
  192 + byteBuf.readBytes(bytes);
  193 + byteBuf.release();
  194 +
  195 + socket.writeBinaryMessage(Buffer.buffer(bytes));
  196 +
  197 + log.info("[sendMessageToDevice][消息已发送到设备，deviceKey: {}，topic: {}，qos: {}，messageId: {}]",
  198 + deviceKey, topic, qos, messageId);
  199 + return true;
  200 + } catch (Exception e) {
  201 + log.error("[sendMessageToDevice][发送消息到设备失败，deviceKey: {}，topic: {}]", deviceKey, topic, e);
  202 + return false;
  203 + }
  204 + }
  205 +
  206 + /**
  207 + * 生成消息 ID
  208 + *
  209 + * @return 消息 ID
  210 + */
  211 + private int generateMessageId() {
  212 + int id = messageIdGenerator.getAndIncrement();
  213 + // MQTT 消息 ID 范围是 1-65535
  214 + if (id > 65535) {
  215 + messageIdGenerator.set(1);
  216 + return 1;
  217 + }
  218 + return id;
  219 + }
  220 +
  221 +}
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/router/IotMqttWsUpstreamHandler.java 0 → 100644
  View file @68437cd
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.mqttws.router;
  2 +
  3 +import cn.hutool.core.util.BooleanUtil;
  4 +import cn.hutool.core.util.IdUtil;
  5 +import cn.hutool.core.util.StrUtil;
  6 +import cn.hutool.extra.spring.SpringUtil;
  7 +import cn.iocoder.yudao.framework.common.pojo.CommonResult;
  8 +import cn.iocoder.yudao.module.iot.core.biz.IotDeviceCommonApi;
  9 +import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceAuthReqDTO;
  10 +import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceGetReqDTO;
  11 +import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceRespDTO;
  12 +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
  13 +import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils;
  14 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqttws.IotMqttWsUpstreamProtocol;
  15 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqttws.manager.IotMqttWsConnectionManager;
  16 +import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
  17 +import io.netty.buffer.ByteBuf;
  18 +import io.netty.buffer.Unpooled;
  19 +import io.netty.channel.embedded.EmbeddedChannel;
  20 +import io.netty.handler.codec.DecoderException;
  21 +import io.netty.handler.codec.mqtt.*;
  22 +import io.vertx.core.buffer.Buffer;
  23 +import io.vertx.core.http.ServerWebSocket;
  24 +import lombok.extern.slf4j.Slf4j;
  25 +
  26 +import java.nio.charset.StandardCharsets;
  27 +import java.util.concurrent.ConcurrentHashMap;
  28 +import java.util.concurrent.atomic.AtomicInteger;
  29 +
  30 +/**
  31 + * IoT MQTT WebSocket 上行消息处理器
  32 + * <p>
  33 + * 处理来自设备的 MQTT 消息，包括：
  34 + * - CONNECT：设备连接认证
  35 + * - PUBLISH：设备发布消息
  36 + * - SUBSCRIBE：设备订阅主题
  37 + * - UNSUBSCRIBE：设备取消订阅
  38 + * - PINGREQ：心跳请求
  39 + * - DISCONNECT：设备断开连接
  40 + *
  41 + * @author 芋道源码
  42 + */
  43 +@Slf4j
  44 +public class IotMqttWsUpstreamHandler {
  45 +
  46 + private final IotMqttWsUpstreamProtocol upstreamProtocol;
  47 +
  48 + private final IotDeviceCommonApi deviceApi;
  49 +
  50 + private final IotDeviceMessageService messageService;
  51 +
  52 + private final IotMqttWsConnectionManager connectionManager;
  53 +
  54 + /**
  55 + * 存储 WebSocket 连接到 Socket ID 的映射
  56 + * Key: WebSocket 对象
  57 + * Value: Socket ID（UUID）
  58 + */
  59 + private final ConcurrentHashMap<ServerWebSocket, String> socketIdMap = new ConcurrentHashMap<>();
  60 +
  61 + /**
  62 + * 存储 Socket ID 对应的设备信息
  63 + * Key: Socket ID（UUID）
  64 + * Value: 设备信息
  65 + */
  66 + private final ConcurrentHashMap<String, IotDeviceRespDTO> socketDeviceMap = new ConcurrentHashMap<>();
  67 +
  68 + /**
  69 + * 存储设备的消息 ID 生成器（用于 QoS > 0 的消息）
  70 + */
  71 + private final ConcurrentHashMap<String, AtomicInteger> deviceMessageIdMap = new ConcurrentHashMap<>();
  72 +
  73 + /**
  74 + * MQTT 解码通道（用于解析 WebSocket 中的 MQTT 二进制消息）
  75 + */
  76 + private final ThreadLocal<EmbeddedChannel> decoderChannelThreadLocal = ThreadLocal
  77 + .withInitial(() -> new EmbeddedChannel(new MqttDecoder()));
  78 +
  79 + /**
  80 + * MQTT 编码通道（用于编码 MQTT 响应消息）
  81 + */
  82 + private final ThreadLocal<EmbeddedChannel> encoderChannelThreadLocal = ThreadLocal
  83 + .withInitial(() -> new EmbeddedChannel(MqttEncoder.INSTANCE));
  84 +
  85 + public IotMqttWsUpstreamHandler(IotMqttWsUpstreamProtocol upstreamProtocol,
  86 + IotDeviceMessageService messageService,
  87 + IotMqttWsConnectionManager connectionManager) {
  88 + this.upstreamProtocol = upstreamProtocol;
  89 + this.deviceApi = SpringUtil.getBean(IotDeviceCommonApi.class);
  90 + this.messageService = messageService;
  91 + this.connectionManager = connectionManager;
  92 + }
  93 +
  94 + /**
  95 + * 处理 WebSocket 连接
  96 + *
  97 + * @param socket WebSocket 连接
  98 + */
  99 + public void handle(ServerWebSocket socket) {
  100 + // 生成唯一的 Socket ID（因为 MQTT 使用二进制协议，textHandlerID() 会返回 null）
  101 + String socketId = IdUtil.simpleUUID();
  102 + socketIdMap.put(socket, socketId);
  103 +
  104 + log.info("[handle][WebSocket 连接建立，socketId: {}，remoteAddress: {}]",
  105 + socketId, socket.remoteAddress());
  106 +
  107 + // 设置二进制数据处理器
  108 + socket.binaryMessageHandler(buffer -> {
  109 + try {
  110 + handleMqttMessage(socket, buffer);
  111 + } catch (Exception e) {
  112 + log.error("[handle][处理 MQTT 消息异常，socketId: {}]", socketId, e);
  113 + socket.close();
  114 + }
  115 + });
  116 +
  117 + // 设置关闭处理器
  118 + socket.closeHandler(v -> {
  119 + socketIdMap.remove(socket);
  120 + IotDeviceRespDTO device = socketDeviceMap.remove(socketId);
  121 + if (device != null) {
  122 + String deviceKey = device.getProductKey() + ":" + device.getDeviceName();
  123 + connectionManager.removeConnection(deviceKey);
  124 + deviceMessageIdMap.remove(deviceKey);
  125 + // 发送设备离线消息
  126 + sendOfflineMessage(device);
  127 + log.info("[handle][WebSocket 连接关闭，deviceKey: {}，socketId: {}]", deviceKey, socketId);
  128 + }
  129 + });
  130 +
  131 + // 设置异常处理器
  132 + socket.exceptionHandler(e -> {
  133 + log.error("[handle][WebSocket 连接异常，socketId: {}]", socketId, e);
  134 + socketIdMap.remove(socket);
  135 + IotDeviceRespDTO device = socketDeviceMap.remove(socketId);
  136 + if (device != null) {
  137 + String deviceKey = device.getProductKey() + ":" + device.getDeviceName();
  138 + connectionManager.removeConnection(deviceKey);
  139 + deviceMessageIdMap.remove(deviceKey);
  140 + }
  141 + socket.close();
  142 + });
  143 + }
  144 +
  145 + /**
  146 + * 处理 MQTT 消息
  147 + *
  148 + * @param socket WebSocket 连接
  149 + * @param buffer 消息缓冲区
  150 + */
  151 + private void handleMqttMessage(ServerWebSocket socket, Buffer buffer) {
  152 + String socketId = socketIdMap.get(socket);
  153 + ByteBuf byteBuf = Unpooled.wrappedBuffer(buffer.getBytes());
  154 +
  155 + try {
  156 + // 使用 EmbeddedChannel 解码 MQTT 消息
  157 + EmbeddedChannel decoderChannel = decoderChannelThreadLocal.get();
  158 + decoderChannel.writeInbound(byteBuf.retain());
  159 +
  160 + // 读取解码后的消息
  161 + MqttMessage mqttMessage = decoderChannel.readInbound();
  162 + if (mqttMessage == null) {
  163 + log.warn("[handleMqttMessage][MQTT 消息解码失败，socketId: {}]", socketId);
  164 + return;
  165 + }
  166 +
  167 + MqttMessageType messageType = mqttMessage.fixedHeader().messageType();
  168 + log.debug("[handleMqttMessage][收到 MQTT 消息，类型: {}，socketId: {}]", messageType, socketId);
  169 +
  170 + // 根据消息类型分发处理
  171 + switch (messageType) {
  172 + case CONNECT:
  173 + handleConnect(socket, (MqttConnectMessage) mqttMessage);
  174 + break;
  175 + case PUBLISH:
  176 + handlePublish(socket, (MqttPublishMessage) mqttMessage);
  177 + break;
  178 + case PUBACK:
  179 + handlePubAck(socket, mqttMessage);
  180 + break;
  181 + case PUBREC:
  182 + handlePubRec(socket, mqttMessage);
  183 + break;
  184 + case PUBREL:
  185 + handlePubRel(socket, mqttMessage);
  186 + break;
  187 + case PUBCOMP:
  188 + handlePubComp(socket, mqttMessage);
  189 + break;
  190 + case SUBSCRIBE:
  191 + handleSubscribe(socket, (MqttSubscribeMessage) mqttMessage);
  192 + break;
  193 + case UNSUBSCRIBE:
  194 + handleUnsubscribe(socket, (MqttUnsubscribeMessage) mqttMessage);
  195 + break;
  196 + case PINGREQ:
  197 + handlePingReq(socket);
  198 + break;
  199 + case DISCONNECT:
  200 + handleDisconnect(socket);
  201 + break;
  202 + default:
  203 + log.warn("[handleMqttMessage][不支持的消息类型: {}，socketId: {}]", messageType, socketId);
  204 + }
  205 + } catch (DecoderException e) {
  206 + log.error("[handleMqttMessage][MQTT 消息解码异常，socketId: {}]", socketId, e);
  207 + socket.close();
  208 + } catch (Exception e) {
  209 + log.error("[handleMqttMessage][处理 MQTT 消息失败，socketId: {}]", socketId, e);
  210 + socket.close();
  211 + } finally {
  212 + byteBuf.release();
  213 + }
  214 + }
  215 +
  216 + /**
  217 + * 处理 CONNECT 消息（设备认证）
  218 + */
  219 + private void handleConnect(ServerWebSocket socket, MqttConnectMessage message) {
  220 + String socketId = socketIdMap.get(socket);
  221 + try {
  222 + // 1. 解析 CONNECT 消息
  223 + MqttConnectPayload payload = message.payload();
  224 + String clientId = payload.clientIdentifier();
  225 + String username = payload.userName();
  226 + String password = payload.passwordInBytes() != null
  227 + ? new String(payload.passwordInBytes(), StandardCharsets.UTF_8)
  228 + : null;
  229 +
  230 + log.info("[handleConnect][收到 CONNECT 消息，clientId: {}，username: {}，socketId: {}]",
  231 + clientId, username, socketId);
  232 +
  233 + // 2. 设备认证
  234 + IotDeviceRespDTO device = authenticateDevice(clientId, username, password);
  235 + if (device == null) {
  236 + log.warn("[handleConnect][设备认证失败，clientId: {}，socketId: {}]", clientId, socketId);
  237 + sendConnAck(socket, MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD);
  238 + socket.close();
  239 + return;
  240 + }
  241 +
  242 + // 3. 保存设备信息
  243 + socketDeviceMap.put(socketId, device);
  244 + String deviceKey = device.getProductKey() + ":" + device.getDeviceName();
  245 + connectionManager.addConnection(deviceKey, socket, socketId);
  246 + deviceMessageIdMap.put(deviceKey, new AtomicInteger(1));
  247 +
  248 + log.info("[handleConnect][设备认证成功，deviceId: {}，deviceKey: {}，socketId: {}]",
  249 + device.getId(), deviceKey, socketId);
  250 +
  251 + // 4. 发送 CONNACK
  252 + sendConnAck(socket, MqttConnectReturnCode.CONNECTION_ACCEPTED);
  253 +
  254 + // 5. 发送设备上线消息
  255 + sendOnlineMessage(device);
  256 +
  257 + } catch (Exception e) {
  258 + log.error("[handleConnect][处理 CONNECT 消息失败，socketId: {}]", socketId, e);
  259 + sendConnAck(socket, MqttConnectReturnCode.CONNECTION_REFUSED_SERVER_UNAVAILABLE);
  260 + socket.close();
  261 + }
  262 + }
  263 +
  264 + /**
  265 + * 处理 PUBLISH 消息（设备发布消息）
  266 + */
  267 + private void handlePublish(ServerWebSocket socket, MqttPublishMessage message) {
  268 + String socketId = socketIdMap.get(socket);
  269 + IotDeviceRespDTO device = socketDeviceMap.get(socketId);
  270 +
  271 + if (device == null) {
  272 + log.warn("[handlePublish][设备未认证，socketId: {}]", socketId);
  273 + socket.close();
  274 + return;
  275 + }
  276 +
  277 + try {
  278 + // 1. 解析 PUBLISH 消息
  279 + MqttFixedHeader fixedHeader = message.fixedHeader();
  280 + MqttPublishVariableHeader variableHeader = message.variableHeader();
  281 + ByteBuf payload = message.payload();
  282 +
  283 + String topic = variableHeader.topicName();
  284 + int messageId = variableHeader.packetId();
  285 + MqttQoS qos = fixedHeader.qosLevel();
  286 +
  287 + log.debug("[handlePublish][收到 PUBLISH 消息，topic: {}，messageId: {}，QoS: {}，deviceId: {}]",
  288 + topic, messageId, qos, device.getId());
  289 +
  290 + // 2. 读取 payload
  291 + byte[] payloadBytes = new byte[payload.readableBytes()];
  292 + payload.readBytes(payloadBytes);
  293 +
  294 + // 3. 解码并发送消息
  295 + IotDeviceMessage deviceMessage = messageService.decodeDeviceMessage(payloadBytes,
  296 + device.getProductKey(), device.getDeviceName());
  297 + if (deviceMessage != null) {
  298 + deviceMessage.setServerId(upstreamProtocol.getServerId());
  299 + messageService.sendDeviceMessage(deviceMessage, device.getProductKey(),
  300 + device.getDeviceName(), upstreamProtocol.getServerId());
  301 + log.info("[handlePublish][设备消息已发送，method: {}，deviceId: {}]",
  302 + deviceMessage.getMethod(), device.getId());
  303 + }
  304 +
  305 + // 4. 根据 QoS 级别发送相应的确认消息
  306 + if (qos == MqttQoS.AT_LEAST_ONCE) {
  307 + // QoS 1：发送 PUBACK
  308 + sendPubAck(socket, messageId);
  309 + } else if (qos == MqttQoS.EXACTLY_ONCE) {
  310 + // QoS 2：发送 PUBREC
  311 + sendPubRec(socket, messageId);
  312 + }
  313 + // QoS 0 无需确认
  314 +
  315 + } catch (Exception e) {
  316 + log.error("[handlePublish][处理 PUBLISH 消息失败，deviceId: {}]", device.getId(), e);
  317 + }
  318 + }
  319 +
  320 + /**
  321 + * 处理 PUBACK 消息（QoS 1 确认）
  322 + */
  323 + private void handlePubAck(ServerWebSocket socket, MqttMessage message) {
  324 + String socketId = socketIdMap.get(socket);
  325 + IotDeviceRespDTO device = socketDeviceMap.get(socketId);
  326 +
  327 + if (device == null) {
  328 + log.warn("[handlePubAck][设备未认证，socketId: {}]", socketId);
  329 + socket.close();
  330 + return;
  331 + }
  332 +
  333 + int messageId = ((MqttMessageIdVariableHeader) message.variableHeader()).messageId();
  334 + log.debug("[handlePubAck][收到 PUBACK，messageId: {}，deviceId: {}]", messageId, device.getId());
  335 + }
  336 +
  337 + /**
  338 + * 处理 PUBREC 消息（QoS 2 第一步确认）
  339 + */
  340 + private void handlePubRec(ServerWebSocket socket, MqttMessage message) {
  341 + String socketId = socketIdMap.get(socket);
  342 + IotDeviceRespDTO device = socketDeviceMap.get(socketId);
  343 +
  344 + if (device == null) {
  345 + log.warn("[handlePubRec][设备未认证，socketId: {}]", socketId);
  346 + socket.close();
  347 + return;
  348 + }
  349 +
  350 + int messageId = ((MqttMessageIdVariableHeader) message.variableHeader()).messageId();
  351 + log.debug("[handlePubRec][收到 PUBREC，messageId: {}，deviceId: {}]", messageId, device.getId());
  352 +
  353 + // 发送 PUBREL
  354 + sendPubRel(socket, messageId);
  355 + }
  356 +
  357 + /**
  358 + * 处理 PUBREL 消息（QoS 2 第二步）
  359 + */
  360 + private void handlePubRel(ServerWebSocket socket, MqttMessage message) {
  361 + String socketId = socketIdMap.get(socket);
  362 + IotDeviceRespDTO device = socketDeviceMap.get(socketId);
  363 +
  364 + if (device == null) {
  365 + log.warn("[handlePubRel][设备未认证，socketId: {}]", socketId);
  366 + socket.close();
  367 + return;
  368 + }
  369 +
  370 + int messageId = ((MqttMessageIdVariableHeader) message.variableHeader()).messageId();
  371 + log.debug("[handlePubRel][收到 PUBREL，messageId: {}，deviceId: {}]", messageId, device.getId());
  372 +
  373 + // 发送 PUBCOMP
  374 + sendPubComp(socket, messageId);
  375 + }
  376 +
  377 + /**
  378 + * 处理 PUBCOMP 消息（QoS 2 完成确认）
  379 + */
  380 + private void handlePubComp(ServerWebSocket socket, MqttMessage message) {
  381 + String socketId = socketIdMap.get(socket);
  382 + IotDeviceRespDTO device = socketDeviceMap.get(socketId);
  383 +
  384 + if (device == null) {
  385 + log.warn("[handlePubComp][设备未认证，socketId: {}]", socketId);
  386 + socket.close();
  387 + return;
  388 + }
  389 +
  390 + int messageId = ((MqttMessageIdVariableHeader) message.variableHeader()).messageId();
  391 + log.debug("[handlePubComp][收到 PUBCOMP，messageId: {}，deviceId: {}]", messageId, device.getId());
  392 + }
  393 +
  394 + /**
  395 + * 处理 SUBSCRIBE 消息（设备订阅主题）
  396 + */
  397 + private void handleSubscribe(ServerWebSocket socket, MqttSubscribeMessage message) {
  398 + String socketId = socketIdMap.get(socket);
  399 + IotDeviceRespDTO device = socketDeviceMap.get(socketId);
  400 +
  401 + if (device == null) {
  402 + log.warn("[handleSubscribe][设备未认证，socketId: {}]", socketId);
  403 + socket.close();
  404 + return;
  405 + }
  406 +
  407 + try {
  408 + // 1. 解析 SUBSCRIBE 消息
  409 + int messageId = message.variableHeader().messageId();
  410 + MqttSubscribePayload payload = message.payload();
  411 + String deviceKey = device.getProductKey() + ":" + device.getDeviceName();
  412 +
  413 + log.info("[handleSubscribe][设备订阅请求，deviceKey: {}，messageId: {}，主题数量: {}]",
  414 + deviceKey, messageId, payload.topicSubscriptions().size());
  415 +
  416 + // 2. 构建 QoS 列表并记录订阅信息
  417 + int[] grantedQosList = new int[payload.topicSubscriptions().size()];
  418 + for (int i = 0; i < payload.topicSubscriptions().size(); i++) {
  419 + MqttTopicSubscription subscription = payload.topicSubscriptions().get(i);
  420 + String topic = subscription.topicFilter();
  421 + grantedQosList[i] = subscription.qualityOfService().value();
  422 +
  423 + // 记录订阅信息到连接管理器
  424 + connectionManager.addSubscription(deviceKey, topic);
  425 +
  426 + log.info("[handleSubscribe][订阅主题: {}，QoS: {}，deviceKey: {}]",
  427 + topic, subscription.qualityOfService(), deviceKey);
  428 + }
  429 +
  430 + // 3. 发送 SUBACK
  431 + sendSubAck(socket, messageId, grantedQosList);
  432 +
  433 + } catch (Exception e) {
  434 + log.error("[handleSubscribe][处理 SUBSCRIBE 消息失败，deviceId: {}]", device.getId(), e);
  435 + }
  436 + }
  437 +
  438 + /**
  439 + * 处理 UNSUBSCRIBE 消息（设备取消订阅）
  440 + */
  441 + private void handleUnsubscribe(ServerWebSocket socket, MqttUnsubscribeMessage message) {
  442 + String socketId = socketIdMap.get(socket);
  443 + IotDeviceRespDTO device = socketDeviceMap.get(socketId);
  444 +
  445 + if (device == null) {
  446 + log.warn("[handleUnsubscribe][设备未认证，socketId: {}]", socketId);
  447 + socket.close();
  448 + return;
  449 + }
  450 +
  451 + try {
  452 + // 1. 解析 UNSUBSCRIBE 消息
  453 + int messageId = message.variableHeader().messageId();
  454 + MqttUnsubscribePayload payload = message.payload();
  455 + String deviceKey = device.getProductKey() + ":" + device.getDeviceName();
  456 +
  457 + log.info("[handleUnsubscribe][设备取消订阅，deviceKey: {}，messageId: {}，主题数量: {}]",
  458 + deviceKey, messageId, payload.topics().size());
  459 +
  460 + // 2. 移除订阅信息
  461 + for (String topic : payload.topics()) {
  462 + connectionManager.removeSubscription(deviceKey, topic);
  463 + log.info("[handleUnsubscribe][取消订阅主题: {}，deviceKey: {}]", topic, deviceKey);
  464 + }
  465 +
  466 + // 3. 发送 UNSUBACK
  467 + sendUnsubAck(socket, messageId);
  468 +
  469 + } catch (Exception e) {
  470 + log.error("[handleUnsubscribe][处理 UNSUBSCRIBE 消息失败，deviceId: {}]", device.getId(), e);
  471 + }
  472 + }
  473 +
  474 + /**
  475 + * 处理 PINGREQ 消息（心跳请求）
  476 + */
  477 + private void handlePingReq(ServerWebSocket socket) {
  478 + String socketId = socketIdMap.get(socket);
  479 + IotDeviceRespDTO device = socketDeviceMap.get(socketId);
  480 +
  481 + if (device == null) {
  482 + log.warn("[handlePingReq][设备未认证，socketId: {}]", socketId);
  483 + socket.close();
  484 + return;
  485 + }
  486 +
  487 + log.debug("[handlePingReq][收到心跳请求，deviceId: {}]", device.getId());
  488 +
  489 + // 发送 PINGRESP
  490 + sendPingResp(socket);
  491 + }
  492 +
  493 + /**
  494 + * 处理 DISCONNECT 消息（设备断开连接）
  495 + */
  496 + private void handleDisconnect(ServerWebSocket socket) {
  497 + String socketId = socketIdMap.get(socket);
  498 + IotDeviceRespDTO device = socketDeviceMap.remove(socketId);
  499 +
  500 + if (device != null) {
  501 + String deviceKey = device.getProductKey() + ":" + device.getDeviceName();
  502 + connectionManager.removeConnection(deviceKey);
  503 + deviceMessageIdMap.remove(deviceKey);
  504 + sendOfflineMessage(device);
  505 + log.info("[handleDisconnect][设备主动断开连接，deviceKey: {}]", deviceKey);
  506 + }
  507 +
  508 + socket.close();
  509 + }
  510 +
  511 + // ==================== 设备认证和状态相关方法 ====================
  512 +
  513 + /**
  514 + * 设备认证
  515 + */
  516 + private IotDeviceRespDTO authenticateDevice(String clientId, String username, String password) {
  517 + try {
  518 + // 1. 参数校验
  519 + if (StrUtil.hasEmpty(clientId, username, password)) {
  520 + log.warn("[authenticateDevice][认证参数不完整，clientId: {}，username: {}]", clientId, username);
  521 + return null;
  522 + }
  523 +
  524 + // 2. 构建认证参数并调用 API
  525 + IotDeviceAuthReqDTO authParams = new IotDeviceAuthReqDTO()
  526 + .setClientId(clientId)
  527 + .setUsername(username)
  528 + .setPassword(password);
  529 +
  530 + CommonResult<Boolean> authResult = deviceApi.authDevice(authParams);
  531 + if (!authResult.isSuccess() || !BooleanUtil.isTrue(authResult.getData())) {
  532 + log.warn("[authenticateDevice][设备认证失败，clientId: {}]", clientId);
  533 + return null;
  534 + }
  535 +
  536 + // 3. 获取设备信息
  537 + IotDeviceAuthUtils.DeviceInfo deviceInfo = IotDeviceAuthUtils.parseUsername(username);
  538 + if (deviceInfo == null) {
  539 + log.warn("[authenticateDevice][用户名格式不正确，username: {}]", username);
  540 + return null;
  541 + }
  542 +
  543 + IotDeviceGetReqDTO getReqDTO = new IotDeviceGetReqDTO()
  544 + .setProductKey(deviceInfo.getProductKey())
  545 + .setDeviceName(deviceInfo.getDeviceName());
  546 +
  547 + CommonResult<IotDeviceRespDTO> deviceResult = deviceApi.getDevice(getReqDTO);
  548 + if (!deviceResult.isSuccess() || deviceResult.getData() == null) {
  549 + log.warn("[authenticateDevice][获取设备信息失败，username: {}]", username);
  550 + return null;
  551 + }
  552 +
  553 + return deviceResult.getData();
  554 + } catch (Exception e) {
  555 + log.error("[authenticateDevice][设备认证异常，clientId: {}]", clientId, e);
  556 + return null;
  557 + }
  558 + }
  559 +
  560 + /**
  561 + * 发送设备上线消息
  562 + */
  563 + private void sendOnlineMessage(IotDeviceRespDTO device) {
  564 + try {
  565 + IotDeviceMessage onlineMessage = IotDeviceMessage.buildStateUpdateOnline();
  566 + messageService.sendDeviceMessage(onlineMessage, device.getProductKey(),
  567 + device.getDeviceName(), upstreamProtocol.getServerId());
  568 + log.info("[sendOnlineMessage][设备上线，deviceId: {}]", device.getId());
  569 + } catch (Exception e) {
  570 + log.error("[sendOnlineMessage][发送设备上线消息失败，deviceId: {}]", device.getId(), e);
  571 + }
  572 + }
  573 +
  574 + /**
  575 + * 发送设备离线消息
  576 + */
  577 + private void sendOfflineMessage(IotDeviceRespDTO device) {
  578 + try {
  579 + IotDeviceMessage offlineMessage = IotDeviceMessage.buildStateOffline();
  580 + messageService.sendDeviceMessage(offlineMessage, device.getProductKey(),
  581 + device.getDeviceName(), upstreamProtocol.getServerId());
  582 + log.info("[sendOfflineMessage][设备离线，deviceId: {}]", device.getId());
  583 + } catch (Exception e) {
  584 + log.error("[sendOfflineMessage][发送设备离线消息失败，deviceId: {}]", device.getId(), e);
  585 + }
  586 + }
  587 +
  588 + // ==================== 发送响应消息的辅助方法 ====================
  589 +
  590 + /**
  591 + * 发送 CONNACK 消息
  592 + */
  593 + private void sendConnAck(ServerWebSocket socket, MqttConnectReturnCode returnCode) {
  594 + try {
  595 + // 构建 CONNACK 消息
  596 + MqttFixedHeader fixedHeader = new MqttFixedHeader(
  597 + MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0);
  598 + MqttConnAckVariableHeader variableHeader = new MqttConnAckVariableHeader(returnCode, false);
  599 + MqttConnAckMessage connAckMessage = new MqttConnAckMessage(fixedHeader, variableHeader);
  600 +
  601 + // 编码并发送
  602 + sendMqttMessage(socket, connAckMessage);
  603 +
  604 + log.debug("[sendConnAck][发送 CONNACK 消息，returnCode: {}]", returnCode);
  605 + } catch (Exception e) {
  606 + log.error("[sendConnAck][发送 CONNACK 消息失败]", e);
  607 + }
  608 + }
  609 +
  610 + /**
  611 + * 发送 PUBACK 消息（QoS 1 确认）
  612 + */
  613 + private void sendPubAck(ServerWebSocket socket, int messageId) {
  614 + try {
  615 + // 构建 PUBACK 消息
  616 + MqttFixedHeader fixedHeader = new MqttFixedHeader(
  617 + MqttMessageType.PUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0);
  618 + MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(messageId);
  619 + MqttMessage pubAckMessage = new MqttMessage(fixedHeader, variableHeader);
  620 +
  621 + // 编码并发送
  622 + sendMqttMessage(socket, pubAckMessage);
  623 +
  624 + log.debug("[sendPubAck][发送 PUBACK 消息，messageId: {}]", messageId);
  625 + } catch (Exception e) {
  626 + log.error("[sendPubAck][发送 PUBACK 消息失败，messageId: {}]", messageId, e);
  627 + }
  628 + }
  629 +
  630 + /**
  631 + * 发送 PUBREC 消息（QoS 2 第一步确认）
  632 + */
  633 + private void sendPubRec(ServerWebSocket socket, int messageId) {
  634 + try {
  635 + // 构建 PUBREC 消息
  636 + MqttFixedHeader fixedHeader = new MqttFixedHeader(
  637 + MqttMessageType.PUBREC, false, MqttQoS.AT_MOST_ONCE, false, 0);
  638 + MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(messageId);
  639 + MqttMessage pubRecMessage = new MqttMessage(fixedHeader, variableHeader);
  640 +
  641 + // 编码并发送
  642 + sendMqttMessage(socket, pubRecMessage);
  643 +
  644 + log.debug("[sendPubRec][发送 PUBREC 消息，messageId: {}]", messageId);
  645 + } catch (Exception e) {
  646 + log.error("[sendPubRec][发送 PUBREC 消息失败，messageId: {}]", messageId, e);
  647 + }
  648 + }
  649 +
  650 + /**
  651 + * 发送 PUBREL 消息（QoS 2 第二步）
  652 + */
  653 + private void sendPubRel(ServerWebSocket socket, int messageId) {
  654 + try {
  655 + // 构建 PUBREL 消息
  656 + MqttFixedHeader fixedHeader = new MqttFixedHeader(
  657 + MqttMessageType.PUBREL, false, MqttQoS.AT_LEAST_ONCE, false, 0);
  658 + MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(messageId);
  659 + MqttMessage pubRelMessage = new MqttMessage(fixedHeader, variableHeader);
  660 +
  661 + // 编码并发送
  662 + sendMqttMessage(socket, pubRelMessage);
  663 +
  664 + log.debug("[sendPubRel][发送 PUBREL 消息，messageId: {}]", messageId);
  665 + } catch (Exception e) {
  666 + log.error("[sendPubRel][发送 PUBREL 消息失败，messageId: {}]", messageId, e);
  667 + }
  668 + }
  669 +
  670 + /**
  671 + * 发送 PUBCOMP 消息（QoS 2 完成确认）
  672 + */
  673 + private void sendPubComp(ServerWebSocket socket, int messageId) {
  674 + try {
  675 + // 构建 PUBCOMP 消息
  676 + MqttFixedHeader fixedHeader = new MqttFixedHeader(
  677 + MqttMessageType.PUBCOMP, false, MqttQoS.AT_MOST_ONCE, false, 0);
  678 + MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(messageId);
  679 + MqttMessage pubCompMessage = new MqttMessage(fixedHeader, variableHeader);
  680 +
  681 + // 编码并发送
  682 + sendMqttMessage(socket, pubCompMessage);
  683 +
  684 + log.debug("[sendPubComp][发送 PUBCOMP 消息，messageId: {}]", messageId);
  685 + } catch (Exception e) {
  686 + log.error("[sendPubComp][发送 PUBCOMP 消息失败，messageId: {}]", messageId, e);
  687 + }
  688 + }
  689 +
  690 + /**
  691 + * 发送 SUBACK 消息
  692 + */
  693 + private void sendSubAck(ServerWebSocket socket, int messageId, int[] grantedQosList) {
  694 + try {
  695 + // 构建 SUBACK 消息
  696 + MqttFixedHeader fixedHeader = new MqttFixedHeader(
  697 + MqttMessageType.SUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0);
  698 + MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(messageId);
  699 + MqttSubAckPayload payload = new MqttSubAckPayload(grantedQosList);
  700 + MqttSubAckMessage subAckMessage = new MqttSubAckMessage(fixedHeader, variableHeader, payload);
  701 +
  702 + // 编码并发送
  703 + sendMqttMessage(socket, subAckMessage);
  704 +
  705 + log.debug("[sendSubAck][发送 SUBACK 消息，messageId: {}，主题数量: {}]", messageId, grantedQosList.length);
  706 + } catch (Exception e) {
  707 + log.error("[sendSubAck][发送 SUBACK 消息失败，messageId: {}]", messageId, e);
  708 + }
  709 + }
  710 +
  711 + /**
  712 + * 发送 UNSUBACK 消息
  713 + */
  714 + private void sendUnsubAck(ServerWebSocket socket, int messageId) {
  715 + try {
  716 + // 构建 UNSUBACK 消息
  717 + MqttFixedHeader fixedHeader = new MqttFixedHeader(
  718 + MqttMessageType.UNSUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0);
  719 + MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(messageId);
  720 + MqttUnsubAckMessage unsubAckMessage = new MqttUnsubAckMessage(fixedHeader, variableHeader);
  721 +
  722 + // 编码并发送
  723 + sendMqttMessage(socket, unsubAckMessage);
  724 +
  725 + log.debug("[sendUnsubAck][发送 UNSUBACK 消息，messageId: {}]", messageId);
  726 + } catch (Exception e) {
  727 + log.error("[sendUnsubAck][发送 UNSUBACK 消息失败，messageId: {}]", messageId, e);
  728 + }
  729 + }
  730 +
  731 + /**
  732 + * 发送 PINGRESP 消息
  733 + */
  734 + private void sendPingResp(ServerWebSocket socket) {
  735 + try {
  736 + // 构建 PINGRESP 消息
  737 + MqttFixedHeader fixedHeader = new MqttFixedHeader(
  738 + MqttMessageType.PINGRESP, false, MqttQoS.AT_MOST_ONCE, false, 0);
  739 + MqttMessage pingRespMessage = new MqttMessage(fixedHeader);
  740 +
  741 + // 编码并发送
  742 + sendMqttMessage(socket, pingRespMessage);
  743 +
  744 + log.debug("[sendPingResp][发送 PINGRESP 消息]");
  745 + } catch (Exception e) {
  746 + log.error("[sendPingResp][发送 PINGRESP 消息失败]", e);
  747 + }
  748 + }
  749 +
  750 + /**
  751 + * 发送 MQTT 消息到 WebSocket
  752 + */
  753 + private void sendMqttMessage(ServerWebSocket socket, MqttMessage mqttMessage) {
  754 + ByteBuf byteBuf = null;
  755 + try {
  756 + // 使用 EmbeddedChannel 编码 MQTT 消息
  757 + EmbeddedChannel encoderChannel = encoderChannelThreadLocal.get();
  758 + encoderChannel.writeOutbound(mqttMessage);
  759 +
  760 + // 读取编码后的 ByteBuf
  761 + byteBuf = encoderChannel.readOutbound();
  762 + if (byteBuf != null) {
  763 + byte[] bytes = new byte[byteBuf.readableBytes()];
  764 + byteBuf.readBytes(bytes);
  765 + socket.writeBinaryMessage(Buffer.buffer(bytes));
  766 + }
  767 + } finally {
  768 + if (byteBuf != null) {
  769 + byteBuf.release();
  770 + }
  771 + }
  772 + }
  773 +
  774 +}
yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application.yaml
  View file @68437cd
@@ -48,7 +48,7 @@ yudao: @@ -48,7 +48,7 @@ yudao:
48 # 针对引入的 HTTP 组件的配置 48 # 针对引入的 HTTP 组件的配置
49 # ==================================== 49 # ====================================
50 http: 50 http:
51 - enabled: true 51 + enabled: false
52 server-port: 8092 52 server-port: 8092
53 # ==================================== 53 # ====================================
54 # 针对引入的 EMQX 组件的配置 54 # 针对引入的 EMQX 组件的配置
@@ -99,11 +99,24 @@ yudao: @@ -99,11 +99,24 @@ yudao:
99 # 针对引入的 MQTT 组件的配置 99 # 针对引入的 MQTT 组件的配置
100 # ==================================== 100 # ====================================
101 mqtt: 101 mqtt:
102 - enabled: true 102 + enabled: false
103 port: 1883 103 port: 1883
104 max-message-size: 8192 104 max-message-size: 8192
105 connect-timeout-seconds: 60 105 connect-timeout-seconds: 60
106 ssl-enabled: false 106 ssl-enabled: false
  107 + # ====================================
  108 + # 针对引入的 MQTT WebSocket 组件的配置
  109 + # ====================================
  110 + mqtt-ws:
  111 + enabled: true # 是否启用 MQTT WebSocket
  112 + port: 8083 # WebSocket 服务端口
  113 + path: /mqtt # WebSocket 路径
  114 + max-message-size: 8192 # 最大消息大小（字节）
  115 + max-frame-size: 65536 # 最大帧大小（字节）
  116 + connect-timeout-seconds: 60 # 连接超时时间（秒）
  117 + keep-alive-timeout-seconds: 300 # 保持连接超时时间（秒）
  118 + ssl-enabled: false # 是否启用 SSL（wss://）
  119 + sub-protocol: mqtt # WebSocket 子协议
107 120
108 --- #################### 日志相关配置 #################### 121 --- #################### 日志相关配置 ####################
109 122
@@ -123,6 +136,7 @@ logging: @@ -123,6 +136,7 @@ logging:
123 cn.iocoder.yudao.module.iot.gateway.protocol.emqx: DEBUG 136 cn.iocoder.yudao.module.iot.gateway.protocol.emqx: DEBUG
124 cn.iocoder.yudao.module.iot.gateway.protocol.http: DEBUG 137 cn.iocoder.yudao.module.iot.gateway.protocol.http: DEBUG
125 cn.iocoder.yudao.module.iot.gateway.protocol.mqtt: DEBUG 138 cn.iocoder.yudao.module.iot.gateway.protocol.mqtt: DEBUG
  139 + cn.iocoder.yudao.module.iot.gateway.protocol.mqttws: DEBUG
126 # 根日志级别 140 # 根日志级别
127 root: INFO 141 root: INFO
128 142
yudao-module-iot/yudao-module-iot-gateway/src/test/resources/mqtt-websocket-test-client.html 0 → 100644
  View file @68437cd
  1 +<!DOCTYPE html>
  2 +<html lang="zh-CN">
  3 +<head>
  4 + <meta charset="UTF-8">
  5 + <meta content="width=device-width, initial-scale=1.0" name="viewport">
  6 + <title>MQTT WebSocket 测试客户端</title>
  7 + <style>
  8 + * {
  9 + margin: 0;
  10 + padding: 0;
  11 + box-sizing: border-box;
  12 + }
  13 +
  14 + body {
  15 + font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif;
  16 + background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
  17 + padding: 20px;
  18 + min-height: 100vh;
  19 + }
  20 +
  21 + .container {
  22 + max-width: 1200px;
  23 + margin: 0 auto;
  24 + background: white;
  25 + border-radius: 10px;
  26 + box-shadow: 0 10px 40px rgba(0, 0, 0, 0.1);
  27 + overflow: hidden;
  28 + }
  29 +
  30 + .header {
  31 + background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
  32 + color: white;
  33 + padding: 30px;
  34 + text-align: center;
  35 + }
  36 +
  37 + .header h1 {
  38 + font-size: 28px;
  39 + margin-bottom: 10px;
  40 + }
  41 +
  42 + .header p {
  43 + opacity: 0.9;
  44 + }
  45 +
  46 + .info-box {
  47 + background: #f8f9fa;
  48 + border-left: 4px solid #667eea;
  49 + padding: 15px;
  50 + margin: 20px;
  51 + border-radius: 5px;
  52 + }
  53 +
  54 + .info-box h3 {
  55 + color: #667eea;
  56 + margin-bottom: 10px;
  57 + font-size: 16px;
  58 + }
  59 +
  60 + .info-box ul {
  61 + margin-left: 20px;
  62 + color: #666;
  63 + font-size: 14px;
  64 + line-height: 1.6;
  65 + }
  66 +
  67 + .info-box code {
  68 + background: #e9ecef;
  69 + padding: 2px 6px;
  70 + border-radius: 3px;
  71 + font-family: 'Courier New', monospace;
  72 + color: #d63384;
  73 + }
  74 +
  75 + .content {
  76 + display: grid;
  77 + grid-template-columns: 1fr 1fr;
  78 + gap: 20px;
  79 + padding: 30px;
  80 + }
  81 +
  82 + .panel {
  83 + background: #f8f9fa;
  84 + border-radius: 8px;
  85 + padding: 20px;
  86 + }
  87 +
  88 + .panel h2 {
  89 + color: #667eea;
  90 + margin-bottom: 20px;
  91 + font-size: 20px;
  92 + border-bottom: 2px solid #667eea;
  93 + padding-bottom: 10px;
  94 + }
  95 +
  96 + .form-group {
  97 + margin-bottom: 15px;
  98 + }
  99 +
  100 + .form-group label {
  101 + display: block;
  102 + margin-bottom: 5px;
  103 + color: #333;
  104 + font-weight: 500;
  105 + }
  106 +
  107 + .form-group input,
  108 + .form-group select,
  109 + .form-group textarea {
  110 + width: 100%;
  111 + padding: 10px;
  112 + border: 1px solid #ddd;
  113 + border-radius: 4px;
  114 + font-size: 14px;
  115 + }
  116 +
  117 + .form-group input:focus,
  118 + .form-group select:focus,
  119 + .form-group textarea:focus {
  120 + outline: none;
  121 + border-color: #667eea;
  122 + }
  123 +
  124 + .form-group textarea {
  125 + resize: vertical;
  126 + min-height: 80px;
  127 + font-family: monospace;
  128 + }
  129 +
  130 + .btn {
  131 + padding: 10px 20px;
  132 + border: none;
  133 + border-radius: 4px;
  134 + cursor: pointer;
  135 + font-size: 14px;
  136 + font-weight: 500;
  137 + transition: all 0.3s;
  138 + }
  139 +
  140 + .btn-primary {
  141 + background: #667eea;
  142 + color: white;
  143 + }
  144 +
  145 + .btn-primary:hover {
  146 + background: #5568d3;
  147 + }
  148 +
  149 + .btn-success {
  150 + background: #28a745;
  151 + color: white;
  152 + }
  153 +
  154 + .btn-success:hover {
  155 + background: #218838;
  156 + }
  157 +
  158 + .btn-danger {
  159 + background: #dc3545;
  160 + color: white;
  161 + }
  162 +
  163 + .btn-danger:hover {
  164 + background: #c82333;
  165 + }
  166 +
  167 + .btn-warning {
  168 + background: #ffc107;
  169 + color: #333;
  170 + }
  171 +
  172 + .btn-warning:hover {
  173 + background: #e0a800;
  174 + }
  175 +
  176 + .btn-group {
  177 + display: flex;
  178 + gap: 10px;
  179 + margin-top: 15px;
  180 + }
  181 +
  182 + .status {
  183 + padding: 10px;
  184 + border-radius: 4px;
  185 + margin-bottom: 20px;
  186 + font-weight: 500;
  187 + }
  188 +
  189 + .status.disconnected {
  190 + background: #f8d7da;
  191 + color: #721c24;
  192 + }
  193 +
  194 + .status.connected {
  195 + background: #d4edda;
  196 + color: #155724;
  197 + }
  198 +
  199 + .status.connecting {
  200 + background: #fff3cd;
  201 + color: #856404;
  202 + }
  203 +
  204 + .log-area {
  205 + background: #1e1e1e;
  206 + color: #d4d4d4;
  207 + padding: 15px;
  208 + border-radius: 4px;
  209 + height: 400px;
  210 + overflow-y: auto;
  211 + font-family: 'Courier New', monospace;
  212 + font-size: 12px;
  213 + line-height: 1.6;
  214 + }
  215 +
  216 + .log-entry {
  217 + margin-bottom: 5px;
  218 + }
  219 +
  220 + .log-entry.info {
  221 + color: #4ec9b0;
  222 + }
  223 +
  224 + .log-entry.success {
  225 + color: #6a9955;
  226 + }
  227 +
  228 + .log-entry.error {
  229 + color: #f48771;
  230 + }
  231 +
  232 + .log-entry.warning {
  233 + color: #dcdcaa;
  234 + }
  235 +
  236 + .stats {
  237 + display: grid;
  238 + grid-template-columns: repeat(3, 1fr);
  239 + gap: 10px;
  240 + margin-top: 15px;
  241 + }
  242 +
  243 + .stat-item {
  244 + background: white;
  245 + padding: 15px;
  246 + border-radius: 4px;
  247 + text-align: center;
  248 + }
  249 +
  250 + .stat-item .value {
  251 + font-size: 24px;
  252 + font-weight: bold;
  253 + color: #667eea;
  254 + }
  255 +
  256 + .stat-item .label {
  257 + font-size: 12px;
  258 + color: #666;
  259 + margin-top: 5px;
  260 + }
  261 +
  262 + @media (max-width: 768px) {
  263 + .content {
  264 + grid-template-columns: 1fr;
  265 + }
  266 + }
  267 + </style>
  268 +</head>
  269 +<body>
  270 +<div class="container">
  271 + <div class="header">
  272 + <h1>🚀 MQTT WebSocket 测试客户端</h1>
  273 + <p>RuoYi-Vue-Pro IoT 模块 - MQTT over WebSocket 在线测试工具</p>
  274 + </div>
  275 +
  276 + <!-- 协议格式说明 -->
  277 + <div class="info-box">
  278 + <h3>📌 标准协议格式说明</h3>
  279 + <ul>
  280 + <li><strong>Topic 格式：</strong><code>/sys/{productKey}/{deviceName}/thing/property/post</code></li>
  281 + <li><strong>Client ID 格式：</strong><code>{productKey}.{deviceName}</code> 例如：<code>zOXKLvHjUqTo7ipD.ceshi001</code>
  282 + </li>
  283 + <li><strong>Username 格式：</strong><code>{deviceName}&{productKey}</code> 例如：<code>ceshi001&zOXKLvHjUqTo7ipD</code>
  284 + </li>
  285 + <li><strong>消息格式（Alink 协议）：</strong>
  286 + <pre style="background: #e9ecef; padding: 10px; border-radius: 5px; margin-top: 5px; overflow-x: auto;">
  287 +{
  288 + "id": "消息 ID（唯一标识）",
  289 + "version": "1.0",
  290 + "method": "thing.property.post",
  291 + "params": {
  292 + "temperature": 25.5,
  293 + "humidity": 60
  294 + }
  295 +}</pre>
  296 + </li>
  297 + <li><strong>常用 Topic（下行 - 服务端推送）：</strong>
  298 + <ul style="margin-top: 5px;">
  299 + <li>属性设置：<code>/sys/{pk}/{dn}/thing/property/set</code></li>
  300 + <li>服务调用：<code>/sys/{pk}/{dn}/thing/service/invoke</code></li>
  301 + <li>配置推送：<code>/sys/{pk}/{dn}/thing/config/push</code></li>
  302 + <li>OTA 升级：<code>/sys/{pk}/{dn}/thing/ota/upgrade</code></li>
  303 + </ul>
  304 + </li>
  305 + <li><strong>常用 Topic（上行 - 设备上报）：</strong>
  306 + <ul style="margin-top: 5px;">
  307 + <li>状态更新：<code>/sys/{pk}/{dn}/thing/state/update</code></li>
  308 + <li>属性上报：<code>/sys/{pk}/{dn}/thing/property/post</code></li>
  309 + <li>事件上报：<code>/sys/{pk}/{dn}/thing/event/post</code></li>
  310 + <li>OTA 进度：<code>/sys/{pk}/{dn}/thing/ota/progress</code></li>
  311 + </ul>
  312 + </li>
  313 + </ul>
  314 + </div>
  315 +
  316 + <div class="content">
  317 + <!-- 连接配置面板 -->
  318 + <div class="panel">
  319 + <h2>📡 连接配置</h2>
  320 +
  321 + <div class="status disconnected" id="statusBar">
  322 + ⚫ 未连接
  323 + </div>
  324 +
  325 + <div class="form-group">
  326 + <label>服务器地址</label>
  327 + <input id="serverUrl" placeholder="ws://host:port/path" type="text" value="ws://localhost:8083/mqtt">
  328 + <small style="color: #666; font-size: 12px;">WebSocket 地址，支持 ws:// 和 wss://</small>
  329 + </div>
  330 +
  331 + <div class="form-group">
  332 + <label>Client ID</label>
  333 + <input id="clientId" placeholder="设备客户端 ID" type="text" value="fqTn4Afs982Nak4N.jiali001">
  334 + <small style="color: #666; font-size: 12px;">格式：{productKey}.{deviceName}</small>
  335 + </div>
  336 +
  337 + <div class="form-group">
  338 + <label>Username</label>
  339 + <input id="username" placeholder="用户名" type="text" value="jiali001&fqTn4Afs982Nak4N">
  340 + <small style="color: #666; font-size: 12px;">格式：{deviceName}&{productKey}</small>
  341 + </div>
  342 +
  343 + <div class="form-group">
  344 + <label>Password</label>
  345 + <input id="password" placeholder="设备密钥"
  346 + type="password" value="ae10188f93febbb6b37bd57f463b2a795ae2800fab8933aef75d3c6422873f28">
  347 + <small style="color: #666; font-size: 12px;">设备的认证密钥（Device Secret）</small>
  348 + </div>
  349 +
  350 + <div class="btn-group">
  351 + <button class="btn btn-success" id="connectBtn" onclick="connect()">🔌 连接</button>
  352 + <button class="btn btn-danger" disabled id="disconnectBtn" onclick="disconnect()">🔌 断开</button>
  353 + <button class="btn btn-warning" onclick="clearLogs()">🗑️ 清空日志</button>
  354 + </div>
  355 +
  356 + <!-- 统计信息 -->
  357 + <div class="stats">
  358 + <div class="stat-item">
  359 + <div class="value" id="sentCount">0</div>
  360 + <div class="label">发送消息数</div>
  361 + </div>
  362 + <div class="stat-item">
  363 + <div class="value" id="receivedCount">0</div>
  364 + <div class="label">接收消息数</div>
  365 + </div>
  366 + <div class="stat-item">
  367 + <div class="value" id="errorCount">0</div>
  368 + <div class="label">错误次数</div>
  369 + </div>
  370 + </div>
  371 + </div>
  372 +
  373 + <!-- 消息发布面板 -->
  374 + <div class="panel">
  375 + <h2>📤 消息发布</h2>
  376 +
  377 + <div class="form-group">
  378 + <label>快捷主题选择（上行消息 - 设备 → 服务端）</label>
  379 + <select id="quickPublishTopicSelect" onchange="selectQuickPublishTopic()"
  380 + style="width: 100%; padding: 8px; border: 1px solid #ddd; border-radius: 5px;">
  381 + <option value="">-- 选择上行消息类型 --</option>
  382 + <option value="thing.state.update">设备状态更新 (thing.state.update)</option>
  383 + <option value="thing.property.post">属性上报 (thing.property.post)</option>
  384 + <option value="thing.event.post">事件上报 (thing.event.post)</option>
  385 + <option value="thing.ota.progress">OTA 升级进度 (thing.ota.progress)</option>
  386 + </select>
  387 + </div>
  388 +
  389 + <div class="form-group">
  390 + <label>主题 (Topic)</label>
  391 + <input id="pubTopic" placeholder="消息主题，格式：/sys/{productKey}/{deviceName}/thing/property/post" type="text"
  392 + value="/sys/fqTn4Afs982Nak4N/jiali001/thing/property/post">
  393 + <small style="color: #666; font-size: 12px;">标准格式：/sys/{productKey}/{deviceName}/thing/property/post</small>
  394 + </div>
  395 +
  396 + <div class="form-group">
  397 + <label>QoS 级别</label>
  398 + <select id="pubQos">
  399 + <option value="0">0 - 最多一次</option>
  400 + <option selected value="1">1 - 至少一次</option>
  401 + <option value="2">2 - 刚好一次</option>
  402 + </select>
  403 + </div>
  404 +
  405 + <div class="form-group">
  406 + <label>消息内容 (JSON - Alink 协议格式)</label>
  407 + <textarea id="pubMessage" placeholder='Alink 协议格式消息'>{
  408 + "id": "123456789",
  409 + "version": "1.0",
  410 + "method": "thing.property.post",
  411 + "params": {
  412 + "temperature": 25.5,
  413 + "humidity": 60
  414 + }
  415 +}</textarea>
  416 + <small style="color: #666; font-size: 12px;">
  417 + Alink 协议格式：id（消息 ID）、version（协议版本）、method（方法）、params（参数）
  418 + </small>
  419 + </div>
  420 +
  421 + <div class="btn-group">
  422 + <button class="btn btn-primary" onclick="publish()">📤 发布消息</button>
  423 + <button class="btn btn-success" onclick="publishSampleData()">📊 发送样例数据</button>
  424 + </div>
  425 +
  426 + <h2 style="margin-top: 30px;">📥 主题订阅</h2>
  427 +
  428 + <div class="form-group">
  429 + <label>快捷主题选择（下行消息 - 服务端 → 设备）</label>
  430 + <select id="quickTopicSelect" onchange="selectQuickTopic()"
  431 + style="width: 100%; padding: 8px; border: 1px solid #ddd; border-radius: 5px;">
  432 + <option value="">-- 选择下行消息类型 --</option>
  433 + <optgroup label="📥 下行消息">
  434 + <option value="thing.property.set">属性设置 (thing.property.set)</option>
  435 + <option value="thing.service.invoke">服务调用 (thing.service.invoke)</option>
  436 + <option value="thing.config.push">配置推送 (thing.config.push)</option>
  437 + <option value="thing.ota.upgrade">OTA 固件推送 (thing.ota.upgrade)</option>
  438 + </optgroup>
  439 + <optgroup label="🔄 回复主题（上行消息的回复）">
  440 + <option value="thing.property.post_reply">属性上报回复 (thing.property.post_reply)</option>
  441 + <option value="thing.event.post_reply">事件上报回复 (thing.event.post_reply)</option>
  442 + </optgroup>
  443 + <optgroup label="🔧 通配符订阅">
  444 + <option value="wildcard_all">订阅所有主题 (/sys/+/+/#)</option>
  445 + <option value="wildcard_thing">订阅所有 thing 主题 (/sys/+/+/thing/#)</option>
  446 + <option value="wildcard_reply">订阅所有回复主题 (/sys/+/+/#_reply)</option>
  447 + </optgroup>
  448 + </select>
  449 + </div>
  450 +
  451 + <div class="form-group">
  452 + <label>订阅主题</label>
  453 + <input id="subTopic" placeholder="订阅主题，格式：/sys/{productKey}/{deviceName}/thing/property/set" type="text"
  454 + value="/sys/fqTn4Afs982Nak4N/jiali001/thing/property/set">
  455 + <small style="color: #666; font-size: 12px;">标准格式：/sys/{productKey}/{deviceName}/thing/method 或使用通配符
  456 + /sys/+/+/#</small>
  457 + </div>
  458 +
  459 + <div class="form-group">
  460 + <label>QoS 级别</label>
  461 + <select id="subQos">
  462 + <option value="0">0 - 最多一次</option>
  463 + <option selected value="1">1 - 至少一次</option>
  464 + <option value="2">2 - 刚好一次</option>
  465 + </select>
  466 + </div>
  467 +
  468 + <div class="btn-group">
  469 + <button class="btn btn-primary" onclick="subscribe()">📥 订阅</button>
  470 + <button class="btn btn-danger" onclick="unsubscribe()">❌ 取消订阅</button>
  471 + </div>
  472 + </div>
  473 +
  474 + <!-- 日志面板 -->
  475 + <div class="panel" style="grid-column: 1 / -1;">
  476 + <h2>📝 日志输出</h2>
  477 + <div class="log-area" id="logArea"></div>
  478 + </div>
  479 + </div>
  480 +</div>
  481 +
  482 +<!-- 使用 MQTT.js 库 -->
  483 +<script src="https://unpkg.com/mqtt/dist/mqtt.min.js"></script>
  484 +
  485 +<script>
  486 + let client = null;
  487 + let sentCount = 0;
  488 + let receivedCount = 0;
  489 + let errorCount = 0;
  490 +
  491 + // 添加日志
  492 + function addLog(message, type = 'info') {
  493 + const logArea = document.getElementById('logArea');
  494 + const timestamp = new Date().toLocaleTimeString();
  495 + const logEntry = document.createElement('div');
  496 + logEntry.className = `log-entry ${type}`;
  497 + logEntry.textContent = `[${timestamp}] ${message}`;
  498 + logArea.appendChild(logEntry);
  499 + logArea.scrollTop = logArea.scrollHeight;
  500 + }
  501 +
  502 + // 更新状态栏
  503 + function updateStatus(status, text) {
  504 + const statusBar = document.getElementById('statusBar');
  505 + statusBar.className = `status ${status}`;
  506 +
  507 + const icons = {
  508 + 'disconnected': '⚫',
  509 + 'connecting': '🟡',
  510 + 'connected': '🟢'
  511 + };
  512 +
  513 + statusBar.textContent = `${icons[status]} ${text}`;
  514 + }
  515 +
  516 + // 更新统计信息
  517 + function updateStats() {
  518 + document.getElementById('sentCount').textContent = sentCount;
  519 + document.getElementById('receivedCount').textContent = receivedCount;
  520 + document.getElementById('errorCount').textContent = errorCount;
  521 + }
  522 +
  523 + // 连接到服务器
  524 + function connect() {
  525 + const serverUrl = document.getElementById('serverUrl').value;
  526 + const clientId = document.getElementById('clientId').value;
  527 + const username = document.getElementById('username').value;
  528 + const password = document.getElementById('password').value;
  529 +
  530 + if (!serverUrl || !clientId) {
  531 + addLog('❌ 请填写服务器地址和 Client ID', 'error');
  532 + errorCount++;
  533 + updateStats();
  534 + return;
  535 + }
  536 +
  537 + updateStatus('connecting', '正在连接...');
  538 + addLog(`🔄 正在连接到 ${serverUrl}...`, 'info');
  539 +
  540 + const options = {
  541 + clientId: clientId,
  542 + username: username,
  543 + password: password,
  544 + clean: true,
  545 + reconnectPeriod: 5000,
  546 + connectTimeout: 30000,
  547 + };
  548 +
  549 + client = mqtt.connect(serverUrl, options);
  550 +
  551 + // 连接成功
  552 + client.on('connect', () => {
  553 + updateStatus('connected', '已连接');
  554 + addLog('✅ 连接成功！', 'success');
  555 + document.getElementById('connectBtn').disabled = true;
  556 + document.getElementById('disconnectBtn').disabled = false;
  557 + });
  558 +
  559 + // 接收消息
  560 + client.on('message', (topic, message) => {
  561 + receivedCount++;
  562 + updateStats();
  563 + addLog(`📥 收到消息 [${topic}]: ${message.toString()}`, 'success');
  564 + });
  565 +
  566 + // 连接错误
  567 + client.on('error', (error) => {
  568 + errorCount++;
  569 + updateStats();
  570 + addLog(`❌ 连接错误: ${error.message}`, 'error');
  571 + });
  572 +
  573 + // 断开连接
  574 + client.on('close', () => {
  575 + updateStatus('disconnected', '未连接');
  576 + addLog('🔌 连接已断开', 'warning');
  577 + document.getElementById('connectBtn').disabled = false;
  578 + document.getElementById('disconnectBtn').disabled = true;
  579 + });
  580 +
  581 + // 离线
  582 + client.on('offline', () => {
  583 + updateStatus('disconnected', '离线');
  584 + addLog('⚠️ 客户端离线', 'warning');
  585 + });
  586 +
  587 + // 重连
  588 + client.on('reconnect', () => {
  589 + updateStatus('connecting', '正在重连...');
  590 + addLog('🔄 正在重连...', 'info');
  591 + });
  592 + }
  593 +
  594 + // 断开连接
  595 + function disconnect() {
  596 + if (client) {
  597 + client.end();
  598 + addLog('👋 主动断开连接', 'info');
  599 + }
  600 + }
  601 +
  602 + // 发布消息
  603 + function publish() {
  604 + if (!client || !client.connected) {
  605 + addLog('❌ 请先连接到服务器', 'error');
  606 + errorCount++;
  607 + updateStats();
  608 + return;
  609 + }
  610 +
  611 + const topic = document.getElementById('pubTopic').value;
  612 + const qos = parseInt(document.getElementById('pubQos').value);
  613 + const message = document.getElementById('pubMessage').value;
  614 +
  615 + if (!topic || !message) {
  616 + addLog('❌ 请填写主题和消息内容', 'error');
  617 + errorCount++;
  618 + updateStats();
  619 + return;
  620 + }
  621 +
  622 + // 验证 JSON 格式
  623 + try {
  624 + JSON.parse(message);
  625 + } catch (e) {
  626 + addLog('⚠️ 消息不是有效的 JSON 格式，将作为纯文本发送', 'warning');
  627 + }
  628 +
  629 + client.publish(topic, message, {qos: qos}, (error) => {
  630 + if (error) {
  631 + errorCount++;
  632 + updateStats();
  633 + addLog(`❌ 发布失败: ${error.message}`, 'error');
  634 + } else {
  635 + sentCount++;
  636 + updateStats();
  637 + addLog(`📤 消息已发布 [${topic}] (QoS ${qos})`, 'success');
  638 + }
  639 + });
  640 + }
  641 +
  642 + // 发送样例数据
  643 + function publishSampleData() {
  644 + // 使用 Alink 协议格式的样例数据
  645 + const sampleData = {
  646 + id: Date.now().toString(),
  647 + version: "1.0",
  648 + method: "thing.property.post",
  649 + params: {
  650 + temperature: parseFloat((20 + Math.random() * 10).toFixed(2)),
  651 + humidity: parseFloat((50 + Math.random() * 20).toFixed(2)),
  652 + pressure: parseFloat((1000 + Math.random() * 50).toFixed(2))
  653 + }
  654 + };
  655 +
  656 + document.getElementById('pubMessage').value = JSON.stringify(sampleData, null, 2);
  657 + addLog('样例数据已生成（Alink 协议格式）', 'info');
  658 + publish();
  659 + }
  660 +
  661 + // 获取 productKey 和 deviceName
  662 + function getDeviceInfo() {
  663 + const clientId = document.getElementById('clientId').value;
  664 + const parts = clientId.split('.');
  665 +
  666 + if (parts.length !== 2) {
  667 + addLog('❌ Client ID 格式不正确（应为 {productKey}.{deviceName}），无法生成主题', 'error');
  668 + return null;
  669 + }
  670 +
  671 + return {
  672 + productKey: parts[0],
  673 + deviceName: parts[1]
  674 + };
  675 + }
  676 +
  677 + // 快捷主题选择（消息发布 - 上行消息）
  678 + function selectQuickPublishTopic() {
  679 + const select = document.getElementById('quickPublishTopicSelect');
  680 + const selectedValue = select.value;
  681 +
  682 + console.log('[selectQuickPublishTopic] 选择的值:', selectedValue);
  683 +
  684 + if (!selectedValue) {
  685 + return;
  686 + }
  687 +
  688 + const deviceInfo = getDeviceInfo();
  689 + if (!deviceInfo) {
  690 + return;
  691 + }
  692 +
  693 + console.log('[selectQuickPublishTopic] 设备信息:', deviceInfo);
  694 +
  695 + // 构建标准主题，将枚举中的点号替换为斜杠
  696 + // 例如：thing.property.post -> /sys/{pk}/{dn}/thing/property/post
  697 + const topic = `/sys/${deviceInfo.productKey}/${deviceInfo.deviceName}/${selectedValue.replace(/\./g, '/')}`;
  698 +
  699 + console.log('[selectQuickPublishTopic] 生成的主题:', topic);
  700 +
  701 + const pubTopicInput = document.getElementById('pubTopic');
  702 + pubTopicInput.value = topic;
  703 +
  704 + console.log('[selectQuickPublishTopic] 输入框的值已设置为:', pubTopicInput.value);
  705 +
  706 + addLog(`📋 已选择发布主题: ${topic}`, 'info');
  707 +
  708 + // 需要 reply 的消息类型（不在 REPLY_DISABLED 列表中）
  709 + const needsReply = [
  710 + 'thing.property.post',
  711 + 'thing.event.post'
  712 + ];
  713 +
  714 + // 如果需要 reply，自动订阅 reply 主题
  715 + if (needsReply.includes(selectedValue)) {
  716 + const replyTopic = `${topic}_reply`;
  717 +
  718 + if (client && client.connected) {
  719 + // 自动订阅 reply 主题
  720 + client.subscribe(replyTopic, {qos: 1}, (err) => {
  721 + if (!err) {
  722 + addLog(`✅ 已自动订阅回复主题: ${replyTopic}`, 'success');
  723 + } else {
  724 + addLog(`❌ 自动订阅回复主题失败: ${err.message}`, 'error');
  725 + }
  726 + });
  727 + } else {
  728 + addLog(`💡 提示: 该消息需要订阅回复主题 ${replyTopic}`, 'warning');
  729 + }
  730 + }
  731 +
  732 + // 重置下拉框到默认选项
  733 + select.selectedIndex = 0;
  734 + console.log('[selectQuickPublishTopic] 下拉框已重置');
  735 + }
  736 +
  737 + // 快捷主题选择（主题订阅 - 下行消息）
  738 + function selectQuickTopic() {
  739 + const select = document.getElementById('quickTopicSelect');
  740 + const selectedValue = select.value;
  741 +
  742 + console.log('[selectQuickTopic] 选择的值:', selectedValue);
  743 +
  744 + if (!selectedValue) {
  745 + return;
  746 + }
  747 +
  748 + const subTopicInput = document.getElementById('subTopic');
  749 +
  750 + // 处理通配符订阅
  751 + if (selectedValue === 'wildcard_all') {
  752 + subTopicInput.value = '/sys/+/+/#';
  753 + addLog('📋 已选择订阅主题: /sys/+/+/#（订阅所有主题）', 'info');
  754 + console.log('[selectQuickTopic] 输入框的值已设置为:', subTopicInput.value);
  755 + select.selectedIndex = 0;
  756 + console.log('[selectQuickTopic] 下拉框已重置');
  757 + return;
  758 + } else if (selectedValue === 'wildcard_thing') {
  759 + subTopicInput.value = '/sys/+/+/thing/#';
  760 + addLog('📋 已选择订阅主题: /sys/+/+/thing/#（订阅所有 thing 主题）', 'info');
  761 + console.log('[selectQuickTopic] 输入框的值已设置为:', subTopicInput.value);
  762 + select.selectedIndex = 0;
  763 + console.log('[selectQuickTopic] 下拉框已重置');
  764 + return;
  765 + } else if (selectedValue === 'wildcard_reply') {
  766 + const deviceInfo = getDeviceInfo();
  767 + if (!deviceInfo) {
  768 + select.selectedIndex = 0;
  769 + return;
  770 + }
  771 + subTopicInput.value = `/sys/${deviceInfo.productKey}/${deviceInfo.deviceName}/#_reply`;
  772 + addLog(`📋 已选择订阅主题: /sys/${deviceInfo.productKey}/${deviceInfo.deviceName}/#_reply（订阅所有回复主题）`, 'info');
  773 + console.log('[selectQuickTopic] 输入框的值已设置为:', subTopicInput.value);
  774 + select.selectedIndex = 0;
  775 + console.log('[selectQuickTopic] 下拉框已重置');
  776 + return;
  777 + }
  778 +
  779 + const deviceInfo = getDeviceInfo();
  780 + if (!deviceInfo) {
  781 + select.selectedIndex = 0;
  782 + return;
  783 + }
  784 +
  785 + console.log('[selectQuickTopic] 设备信息:', deviceInfo);
  786 +
  787 + // 构建标准主题，将枚举中的点号替换为斜杠
  788 + // 例如：thing.property.set -> /sys/{pk}/{dn}/thing/property/set
  789 + const topic = `/sys/${deviceInfo.productKey}/${deviceInfo.deviceName}/${selectedValue.replace(/\./g, '/')}`;
  790 +
  791 + console.log('[selectQuickTopic] 生成的主题:', topic);
  792 +
  793 + subTopicInput.value = topic;
  794 + addLog(`📋 已选择订阅主题: ${topic}`, 'info');
  795 +
  796 + console.log('[selectQuickTopic] 输入框的值已设置为:', subTopicInput.value);
  797 +
  798 + // 重置下拉框到默认选项
  799 + select.selectedIndex = 0;
  800 + console.log('[selectQuickTopic] 下拉框已重置');
  801 + }
  802 +
  803 + // 订阅主题
  804 + function subscribe() {
  805 + if (!client || !client.connected) {
  806 + addLog('❌ 请先连接到服务器', 'error');
  807 + errorCount++;
  808 + updateStats();
  809 + return;
  810 + }
  811 +
  812 + const topic = document.getElementById('subTopic').value;
  813 + const qos = parseInt(document.getElementById('subQos').value);
  814 +
  815 + if (!topic) {
  816 + addLog('❌ 请填写订阅主题', 'error');
  817 + errorCount++;
  818 + updateStats();
  819 + return;
  820 + }
  821 +
  822 + client.subscribe(topic, {qos: qos}, (error) => {
  823 + if (error) {
  824 + errorCount++;
  825 + updateStats();
  826 + addLog(`❌ 订阅失败: ${error.message}`, 'error');
  827 + } else {
  828 + addLog(`📥 已订阅主题 [${topic}] (QoS ${qos})`, 'success');
  829 + }
  830 + });
  831 + }
  832 +
  833 + // 取消订阅
  834 + function unsubscribe() {
  835 + if (!client || !client.connected) {
  836 + addLog('❌ 请先连接到服务器', 'error');
  837 + errorCount++;
  838 + updateStats();
  839 + return;
  840 + }
  841 +
  842 + const topic = document.getElementById('subTopic').value;
  843 +
  844 + if (!topic) {
  845 + addLog('❌ 请填写要取消的订阅主题', 'error');
  846 + errorCount++;
  847 + updateStats();
  848 + return;
  849 + }
  850 +
  851 + client.unsubscribe(topic, (error) => {
  852 + if (error) {
  853 + errorCount++;
  854 + updateStats();
  855 + addLog(`❌ 取消订阅失败: ${error.message}`, 'error');
  856 + } else {
  857 + addLog(`❌ 已取消订阅 [${topic}]`, 'info');
  858 + }
  859 + });
  860 + }
  861 +
  862 + // 清空日志
  863 + function clearLogs() {
  864 + document.getElementById('logArea').innerHTML = '';
  865 + sentCount = 0;
  866 + receivedCount = 0;
  867 + errorCount = 0;
  868 + updateStats();
  869 + addLog('🗑️ 日志已清空', 'info');
  870 + }
  871 +
  872 + // 页面加载完成
  873 + window.onload = function () {
  874 + addLog('👋 欢迎使用 MQTT WebSocket 测试客户端！', 'success');
  875 + addLog('📚 请配置连接参数后点击"连接"按钮', 'info');
  876 + };
  877 +
  878 + // 页面关闭前断开连接
  879 + window.onbeforeunload = function () {
  880 + if (client && client.connected) {
  881 + client.end();
  882 + }
  883 + };
  884 +</script>
  885 +</body>
  886 +</html>
  887 +
  888 +