From 68437cd8308307c1b5a84475c64443dd0b69016a Mon Sep 17 00:00:00 2001 From: haohao <1036606149@qq.com> Date: Thu, 9 Oct 2025 22:27:30 +0800 Subject: [PATCH] feat：【IoT 物联网】新增 MQTT WebSocket 协议支持 --- yudao-dependencies/pom.xml | 2 +- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayConfiguration.java | 42 ++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayProperties.java | 101 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/IotMqttWsDownstreamSubscriber.java | 79 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/IotMqttWsUpstreamProtocol.java | 146 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/manager/IotMqttWsConnectionManager.java | 257 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/package-info.java | 15 +++++++++++++++ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/router/IotMqttWsDownstreamHandler.java | 221 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/router/IotMqttWsUpstreamHandler.java | 774 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application.yaml | 18 ++++++++++++++++-- yudao-module-iot/yudao-module-iot-gateway/src/test/resources/mqtt-websocket-test-client.html | 888 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 11 files changed, 2540 insertions(+), 3 deletions(-) create mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/IotMqttWsDownstreamSubscriber.java create mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/IotMqttWsUpstreamProtocol.java create mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/manager/IotMqttWsConnectionManager.java create mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/package-info.java create mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/router/IotMqttWsDownstreamHandler.java create mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/router/IotMqttWsUpstreamHandler.java create mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/test/resources/mqtt-websocket-test-client.html diff --git a/yudao-dependencies/pom.xml b/yudao-dependencies/pom.xml index e08f834..ada6130 100644 --- a/yudao-dependencies/pom.xml +++ b/yudao-dependencies/pom.xml @@ -68,7 +68,7 @@ 4.2.4.Final 1.2.5 0.9.0 - 4.5.13 + 4.5.21 2.30.14 1.16.7 diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayConfiguration.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayConfiguration.java index 4b9c3af..fab4c8c 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayConfiguration.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayConfiguration.java @@ -10,6 +10,10 @@ import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttDownstreamSubscr import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttUpstreamProtocol; import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.manager.IotMqttConnectionManager; import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router.IotMqttDownstreamHandler; +import cn.iocoder.yudao.module.iot.gateway.protocol.mqttws.IotMqttWsDownstreamSubscriber; +import cn.iocoder.yudao.module.iot.gateway.protocol.mqttws.IotMqttWsUpstreamProtocol; +import cn.iocoder.yudao.module.iot.gateway.protocol.mqttws.manager.IotMqttWsConnectionManager; +import cn.iocoder.yudao.module.iot.gateway.protocol.mqttws.router.IotMqttWsDownstreamHandler; import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.IotTcpDownstreamSubscriber; import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.IotTcpUpstreamProtocol; import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.manager.IotTcpConnectionManager; @@ -151,4 +155,42 @@ public class IotGatewayConfiguration { } + /** + * IoT 网关 MQTT WebSocket 协议配置类 + */ + @Configuration + @ConditionalOnProperty(prefix = "yudao.iot.gateway.protocol.mqtt-ws", name = "enabled", havingValue = "true") + @Slf4j + public static class MqttWsProtocolConfiguration { + + @Bean(destroyMethod = "close") + public Vertx mqttWsVertx() { + return Vertx.vertx(); + } + + @Bean + public IotMqttWsUpstreamProtocol iotMqttWsUpstreamProtocol(IotGatewayProperties gatewayProperties, + IotDeviceMessageService messageService, + IotMqttWsConnectionManager connectionManager, + Vertx mqttWsVertx) { + return new IotMqttWsUpstreamProtocol(gatewayProperties.getProtocol().getMqttWs(), + messageService, connectionManager, mqttWsVertx); + } + + @Bean + public IotMqttWsDownstreamHandler iotMqttWsDownstreamHandler(IotDeviceMessageService messageService, + IotDeviceService deviceService, + IotMqttWsConnectionManager connectionManager) { + return new IotMqttWsDownstreamHandler(messageService, deviceService, connectionManager); + } + + @Bean + public IotMqttWsDownstreamSubscriber iotMqttWsDownstreamSubscriber(IotMqttWsUpstreamProtocol mqttWsUpstreamProtocol, + IotMqttWsDownstreamHandler downstreamHandler, + IotMessageBus messageBus) { + return new IotMqttWsDownstreamSubscriber(mqttWsUpstreamProtocol, downstreamHandler, messageBus); + } + + } + } diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayProperties.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayProperties.java index 2c2000f..7655a37 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayProperties.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayProperties.java @@ -88,6 +88,11 @@ public class IotGatewayProperties { */ private MqttProperties mqtt; + /** + * MQTT WebSocket 组件配置 + */ + private MqttWsProperties mqttWs; + } @Data @@ -402,4 +407,100 @@ public class IotGatewayProperties { } + @Data + public static class MqttWsProperties { + + /** + * 是否开启 + */ + @NotNull(message = "是否开启不能为空") + private Boolean enabled; + + /** + * WebSocket 服务器端口（默认：8083） + */ + private Integer port = 8083; + + /** + * WebSocket 路径（默认：/mqtt） + */ + @NotEmpty(message = "WebSocket 路径不能为空") + private String path = "/mqtt"; + + /** + * 最大消息大小（字节） + */ + private Integer maxMessageSize = 8192; + + /** + * 连接超时时间（秒） + */ + private Integer connectTimeoutSeconds = 60; + + /** + * 保持连接超时时间（秒） + */ + private Integer keepAliveTimeoutSeconds = 300; + + /** + * 是否启用 SSL（wss://） + */ + private Boolean sslEnabled = false; + + /** + * SSL 配置 + */ + private SslOptions sslOptions = new SslOptions(); + + /** + * WebSocket 子协议（通常为 "mqtt" 或 "mqttv3.1"） + */ + @NotEmpty(message = "WebSocket 子协议不能为空") + private String subProtocol = "mqtt"; + + /** + * 最大帧大小（字节） + */ + private Integer maxFrameSize = 65536; + + /** + * SSL 配置选项 + */ + @Data + public static class SslOptions { + + /** + * 密钥证书选项 + */ + private io.vertx.core.net.KeyCertOptions keyCertOptions; + + /** + * 信任选项 + */ + private io.vertx.core.net.TrustOptions trustOptions; + + /** + * SSL 证书路径 + */ + private String certPath; + + /** + * SSL 私钥路径 + */ + private String keyPath; + + /** + * 信任存储路径 + */ + private String trustStorePath; + + /** + * 信任存储密码 + */ + private String trustStorePassword; + + } + + } + } diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/IotMqttWsDownstreamSubscriber.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/IotMqttWsDownstreamSubscriber.java new file mode 100644 index 0000000..302824d --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/IotMqttWsDownstreamSubscriber.java @@ -0,0 +1,79 @@ +package cn.iocoder.yudao.module.iot.gateway.protocol.mqttws; + +import cn.hutool.core.util.StrUtil; +import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus; +import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber; +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; +import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils; +import cn.iocoder.yudao.module.iot.gateway.protocol.mqttws.router.IotMqttWsDownstreamHandler; +import jakarta.annotation.PostConstruct; +import lombok.extern.slf4j.Slf4j; + +/** + * IoT MQTT WebSocket 下行消息订阅器 + *

+ * 订阅消息总线的设备下行消息，并通过 WebSocket 发送到设备 + * + * @author 芋道源码 + */ +@Slf4j +public class IotMqttWsDownstreamSubscriber implements IotMessageSubscriber { + + private final IotMqttWsUpstreamProtocol upstreamProtocol; + private final IotMqttWsDownstreamHandler downstreamHandler; + private final IotMessageBus messageBus; + + public IotMqttWsDownstreamSubscriber(IotMqttWsUpstreamProtocol upstreamProtocol, + IotMqttWsDownstreamHandler downstreamHandler, + IotMessageBus messageBus) { + this.upstreamProtocol = upstreamProtocol; + this.downstreamHandler = downstreamHandler; + this.messageBus = messageBus; + } + + @PostConstruct + public void init() { + messageBus.register(this); + log.info("[init][MQTT WebSocket 下行消息订阅器已启动，topic: {}]", getTopic()); + } + + @Override + public String getTopic() { + return IotDeviceMessageUtils.buildMessageBusGatewayDeviceMessageTopic(upstreamProtocol.getServerId()); + } + + @Override + public String getGroup() { + // 保证点对点消费，需要保证独立的 Group，所以使用 Topic 作为 Group + return getTopic(); + } + + @Override + public void onMessage(IotDeviceMessage message) { + log.debug("[onMessage][收到下行消息，deviceId: {}，method: {}]", + message.getDeviceId(), message.getMethod()); + try { + // 1. 校验 + String method = message.getMethod(); + if (StrUtil.isBlank(method)) { + log.warn("[onMessage][消息方法为空，deviceId: {}]", message.getDeviceId()); + return; + } + + // 2. 委托给下行处理器处理业务逻辑 + boolean success = downstreamHandler.handleDownstreamMessage(message); + if (success) { + log.debug("[onMessage][下行消息处理成功，deviceId: {}，method: {}]", + message.getDeviceId(), message.getMethod()); + } else { + log.warn("[onMessage][下行消息处理失败，deviceId: {}，method: {}]", + message.getDeviceId(), message.getMethod()); + } + } catch (Exception e) { + log.error("[onMessage][处理下行消息失败，deviceId: {}，method: {}]", + message.getDeviceId(), message.getMethod(), e); + } + } + +} + diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/IotMqttWsUpstreamProtocol.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/IotMqttWsUpstreamProtocol.java new file mode 100644 index 0000000..6944d47 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/IotMqttWsUpstreamProtocol.java @@ -0,0 +1,146 @@ +package cn.iocoder.yudao.module.iot.gateway.protocol.mqttws; + +import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils; +import cn.iocoder.yudao.module.iot.gateway.config.IotGatewayProperties; +import cn.iocoder.yudao.module.iot.gateway.protocol.mqttws.manager.IotMqttWsConnectionManager; +import cn.iocoder.yudao.module.iot.gateway.protocol.mqttws.router.IotMqttWsUpstreamHandler; +import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService; +import io.vertx.core.Vertx; +import io.vertx.core.http.HttpServer; +import io.vertx.core.http.HttpServerOptions; +import io.vertx.core.http.ServerWebSocket; +import jakarta.annotation.PostConstruct; +import jakarta.annotation.PreDestroy; +import lombok.Getter; +import lombok.extern.slf4j.Slf4j; + +/** + * IoT 网关 MQTT WebSocket 协议：接收设备上行消息 + *

+ * 基于 Vert.x 实现 MQTT over WebSocket 服务端，支持： + * - 标准 MQTT 3.1.1 协议 + * - WebSocket 协议升级 + * - SSL/TLS 加密（wss://） + * - 设备认证与连接管理 + * - QoS 0/1/2 消息质量保证 + * + * @author 芋道源码 + */ +@Slf4j +public class IotMqttWsUpstreamProtocol { + + private final IotGatewayProperties.MqttWsProperties mqttWsProperties; + + private final IotDeviceMessageService messageService; + + private final IotMqttWsConnectionManager connectionManager; + + private final Vertx vertx; + + @Getter + private final String serverId; + + private HttpServer httpServer; + + public IotMqttWsUpstreamProtocol(IotGatewayProperties.MqttWsProperties mqttWsProperties, + IotDeviceMessageService messageService, + IotMqttWsConnectionManager connectionManager, + Vertx vertx) { + this.mqttWsProperties = mqttWsProperties; + this.messageService = messageService; + this.connectionManager = connectionManager; + this.vertx = vertx; + this.serverId = IotDeviceMessageUtils.generateServerId(mqttWsProperties.getPort()); + } + + @PostConstruct + public void start() { + // 创建 HTTP 服务器选项 + HttpServerOptions options = new HttpServerOptions() + .setPort(mqttWsProperties.getPort()) + .setIdleTimeout(mqttWsProperties.getKeepAliveTimeoutSeconds()) + .setMaxWebSocketFrameSize(mqttWsProperties.getMaxFrameSize()) + .setMaxWebSocketMessageSize(mqttWsProperties.getMaxMessageSize()) + // 配置 WebSocket 子协议支持 + .addWebSocketSubProtocol(mqttWsProperties.getSubProtocol()); + + // 配置 SSL（如果启用） + if (Boolean.TRUE.equals(mqttWsProperties.getSslEnabled())) { + options.setSsl(true) + .setKeyCertOptions(mqttWsProperties.getSslOptions().getKeyCertOptions()) + .setTrustOptions(mqttWsProperties.getSslOptions().getTrustOptions()); + log.info("[start][MQTT WebSocket 已启用 SSL/TLS (wss://)]"); + } + + // 创建 HTTP 服务器 + httpServer = vertx.createHttpServer(options); + + // 设置 WebSocket 处理器 + httpServer.webSocketHandler(this::handleWebSocketConnection); + + // 启动服务器 + try { + httpServer.listen().result(); + log.info("[start][IoT 网关 MQTT WebSocket 协议启动成功，端口: {}，路径: {}，支持子协议: {}]", + mqttWsProperties.getPort(), mqttWsProperties.getPath(), + "mqtt, mqttv3.1, " + mqttWsProperties.getSubProtocol()); + } catch (Exception e) { + log.error("[start][IoT 网关 MQTT WebSocket 协议启动失败]", e); + throw e; + } + } + + @PreDestroy + public void stop() { + if (httpServer != null) { + try { + // 关闭所有连接 + connectionManager.closeAllConnections(); + + // 关闭服务器 + httpServer.close().result(); + log.info("[stop][IoT 网关 MQTT WebSocket 协议已停止]"); + } catch (Exception e) { + log.error("[stop][IoT 网关 MQTT WebSocket 协议停止失败]", e); + } + } + } + + /** + * 处理 WebSocket 连接请求 + * + * @param socket WebSocket 连接 + */ + private void handleWebSocketConnection(ServerWebSocket socket) { + String path = socket.path(); + String subProtocol = socket.subProtocol(); + + log.info("[handleWebSocketConnection][收到 WebSocket 连接请求，path: {}，subProtocol: {}，remoteAddress: {}]", + path, subProtocol, socket.remoteAddress()); + + // 验证路径 + if (!mqttWsProperties.getPath().equals(path)) { + log.warn("[handleWebSocketConnection][WebSocket 路径不匹配，拒绝连接，path: {}，期望: {}]", + path, mqttWsProperties.getPath()); + socket.close(); + return; + } + + // 验证子协议 + // Vert.x 已经自动进行了子协议协商，这里只需要验证是否为 MQTT 相关协议 + if (subProtocol != null && !subProtocol.startsWith("mqtt")) { + log.warn("[handleWebSocketConnection][WebSocket 子协议不支持，拒绝连接，subProtocol: {}]", subProtocol); + socket.close(); + return; + } + + log.info("[handleWebSocketConnection][WebSocket 连接已接受，remoteAddress: {}，subProtocol: {}]", + socket.remoteAddress(), subProtocol); + + // 创建处理器并处理连接 + IotMqttWsUpstreamHandler handler = new IotMqttWsUpstreamHandler( + this, messageService, connectionManager); + handler.handle(socket); + } + +} diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/manager/IotMqttWsConnectionManager.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/manager/IotMqttWsConnectionManager.java new file mode 100644 index 0000000..2bf61be --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/manager/IotMqttWsConnectionManager.java @@ -0,0 +1,257 @@ +package cn.iocoder.yudao.module.iot.gateway.protocol.mqttws.manager; + +import io.vertx.core.http.ServerWebSocket; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; + +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArraySet; + +/** + * IoT MQTT WebSocket 连接管理器 + * + * @author 芋道源码 + */ +@Slf4j +@Component +public class IotMqttWsConnectionManager { + + /** + * 存储设备连接 + * Key: 设备标识（deviceKey） + * Value: WebSocket 连接 + */ + private final Map connections = new ConcurrentHashMap<>(); + + /** + * 存储设备标识与 Socket ID 的映射 + * Key: 设备标识（deviceKey） + * Value: Socket ID（UUID） + */ + private final Map deviceKeyToSocketId = new ConcurrentHashMap<>(); + + /** + * 存储 Socket ID 与设备标识的映射 + * Key: Socket ID（UUID） + * Value: 设备标识（deviceKey） + */ + private final Map socketIdToDeviceKey = new ConcurrentHashMap<>(); + + /** + * 存储设备订阅的主题 + * Key: 设备标识（deviceKey） + * Value: 订阅的主题集合 + */ + private final Map> deviceSubscriptions = new ConcurrentHashMap<>(); + + /** + * 添加连接 + * + * @param deviceKey 设备标识 + * @param socket WebSocket 连接 + * @param socketId Socket ID（UUID） + */ + public void addConnection(String deviceKey, ServerWebSocket socket, String socketId) { + connections.put(deviceKey, socket); + deviceKeyToSocketId.put(deviceKey, socketId); + socketIdToDeviceKey.put(socketId, deviceKey); + log.info("[addConnection][设备连接已添加，deviceKey: {}，socketId: {}，当前连接数: {}]", + deviceKey, socketId, connections.size()); + } + + /** + * 移除连接 + * + * @param deviceKey 设备标识 + */ + public void removeConnection(String deviceKey) { + ServerWebSocket socket = connections.remove(deviceKey); + String socketId = deviceKeyToSocketId.remove(deviceKey); + if (socketId != null) { + socketIdToDeviceKey.remove(socketId); + } + if (socket != null) { + log.info("[removeConnection][设备连接已移除，deviceKey: {}，socketId: {}，当前连接数: {}]", + deviceKey, socketId, connections.size()); + } + } + + /** + * 根据 Socket ID 移除连接 + * + * @param socketId WebSocket 文本框架 ID + */ + public void removeConnectionBySocketId(String socketId) { + String deviceKey = socketIdToDeviceKey.remove(socketId); + if (deviceKey != null) { + connections.remove(deviceKey); + log.info("[removeConnectionBySocketId][设备连接已移除，socketId: {}，deviceKey: {}，当前连接数: {}]", + socketId, deviceKey, connections.size()); + } + } + + /** + * 获取连接 + * + * @param deviceKey 设备标识 + * @return WebSocket 连接 + */ + public ServerWebSocket getConnection(String deviceKey) { + return connections.get(deviceKey); + } + + /** + * 根据 Socket ID 获取设备标识 + * + * @param socketId WebSocket 文本框架 ID + * @return 设备标识 + */ + public String getDeviceKeyBySocketId(String socketId) { + return socketIdToDeviceKey.get(socketId); + } + + /** + * 检查设备是否在线 + * + * @param deviceKey 设备标识 + * @return 是否在线 + */ + public boolean isOnline(String deviceKey) { + return connections.containsKey(deviceKey); + } + + /** + * 获取当前连接数 + * + * @return 连接数 + */ + public int getConnectionCount() { + return connections.size(); + } + + /** + * 关闭所有连接 + */ + public void closeAllConnections() { + connections.forEach((deviceKey, socket) -> { + try { + socket.close(); + log.info("[closeAllConnections][关闭设备连接，deviceKey: {}]", deviceKey); + } catch (Exception e) { + log.error("[closeAllConnections][关闭设备连接失败，deviceKey: {}]", deviceKey, e); + } + }); + connections.clear(); + deviceKeyToSocketId.clear(); + socketIdToDeviceKey.clear(); + deviceSubscriptions.clear(); + log.info("[closeAllConnections][所有连接已关闭]"); + } + + // ==================== 订阅管理方法 ==================== + + /** + * 添加订阅 + * + * @param deviceKey 设备标识 + * @param topic 订阅主题 + */ + public void addSubscription(String deviceKey, String topic) { + deviceSubscriptions.computeIfAbsent(deviceKey, k -> new CopyOnWriteArraySet<>()).add(topic); + log.debug("[addSubscription][设备订阅主题，deviceKey: {}，topic: {}]", deviceKey, topic); + } + + /** + * 移除订阅 + * + * @param deviceKey 设备标识 + * @param topic 订阅主题 + */ + public void removeSubscription(String deviceKey, String topic) { + Set topics = deviceSubscriptions.get(deviceKey); + if (topics != null) { + topics.remove(topic); + log.debug("[removeSubscription][设备取消订阅，deviceKey: {}，topic: {}]", deviceKey, topic); + } + } + + /** + * 检查设备是否订阅了指定主题 + * 支持 MQTT 通配符匹配（+ 和 #） + * + * @param deviceKey 设备标识 + * @param topic 发布主题 + * @return 是否匹配 + */ + public boolean isSubscribed(String deviceKey, String topic) { + Set subscriptions = deviceSubscriptions.get(deviceKey); + if (subscriptions == null || subscriptions.isEmpty()) { + return false; + } + + // 检查是否有匹配的订阅 + for (String subscription : subscriptions) { + if (topicMatches(subscription, topic)) { + return true; + } + } + return false; + } + + /** + * 获取设备的所有订阅 + * + * @param deviceKey 设备标识 + * @return 订阅主题集合 + */ + public Set getSubscriptions(String deviceKey) { + return deviceSubscriptions.get(deviceKey); + } + + /** + * MQTT 主题匹配 + * 支持通配符： + * - +：匹配单层主题 + * - #：匹配多层主题（必须在末尾） + * + * @param subscription 订阅主题（可能包含通配符） + * @param topic 发布主题（不包含通配符） + * @return 是否匹配 + */ + private boolean topicMatches(String subscription, String topic) { + // 完全匹配 + if (subscription.equals(topic)) { + return true; + } + + // 不包含通配符 + if (!subscription.contains("+") && !subscription.contains("#")) { + return false; + } + + String[] subscriptionParts = subscription.split("/"); + String[] topicParts = topic.split("/"); + + int i = 0; + for (; i < subscriptionParts.length && i < topicParts.length; i++) { + String subPart = subscriptionParts[i]; + String topicPart = topicParts[i]; + + if (subPart.equals("#")) { + // # 匹配剩余所有层级，且必须在末尾 + return i == subscriptionParts.length - 1; + } + + if (!subPart.equals("+") && !subPart.equals(topicPart)) { + // 不是通配符且不匹配 + return false; + } + } + + // 检查是否都匹配完 + return i == subscriptionParts.length && i == topicParts.length; + } + +} diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/package-info.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/package-info.java new file mode 100644 index 0000000..b9af4af --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/package-info.java @@ -0,0 +1,15 @@ +/** + * IoT 网关 MQTT WebSocket 协议实现 + *

+ * 基于 Vert.x 实现 MQTT over WebSocket 服务端，支持： + * - 标准 MQTT 3.1.1 协议 + * - WebSocket 协议升级 + * - SSL/TLS 加密（wss://） + * - 设备认证与连接管理 + * - QoS 0/1/2 消息质量保证 + * - 双向消息通信（上行/下行） + * + * @author 芋道源码 + */ +package cn.iocoder.yudao.module.iot.gateway.protocol.mqttws; + diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/router/IotMqttWsDownstreamHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/router/IotMqttWsDownstreamHandler.java new file mode 100644 index 0000000..37148de --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/router/IotMqttWsDownstreamHandler.java @@ -0,0 +1,221 @@ +package cn.iocoder.yudao.module.iot.gateway.protocol.mqttws.router; + +import cn.hutool.core.util.StrUtil; +import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceRespDTO; +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; +import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils; +import cn.iocoder.yudao.module.iot.gateway.protocol.mqttws.manager.IotMqttWsConnectionManager; +import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceService; +import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService; +import cn.iocoder.yudao.module.iot.gateway.util.IotMqttTopicUtils; +import io.vertx.core.buffer.Buffer; +import io.vertx.core.http.ServerWebSocket; +import lombok.extern.slf4j.Slf4j; + +import java.util.concurrent.atomic.AtomicInteger; + +/** + * IoT MQTT WebSocket 下行消息处理器 + *

+ * 处理从消息总线发送到设备的消息，包括： + * - 属性设置 + * - 服务调用 + * - 事件通知 + * + * @author 芋道源码 + */ +@Slf4j +public class IotMqttWsDownstreamHandler { + + private final IotDeviceMessageService deviceMessageService; + + private final IotDeviceService deviceService; + + private final IotMqttWsConnectionManager connectionManager; + + /** + * 消息 ID 生成器（用于发布消息） + */ + private final AtomicInteger messageIdGenerator = new AtomicInteger(1); + + public IotMqttWsDownstreamHandler(IotDeviceMessageService deviceMessageService, + IotDeviceService deviceService, + IotMqttWsConnectionManager connectionManager) { + this.deviceMessageService = deviceMessageService; + this.deviceService = deviceService; + this.connectionManager = connectionManager; + } + + /** + * 处理下行消息 + * + * @param message 设备消息 + * @return 是否处理成功 + */ + public boolean handleDownstreamMessage(IotDeviceMessage message) { + try { + // 1. 基础校验 + if (message == null || message.getDeviceId() == null) { + log.warn("[handleDownstreamMessage][消息或设备 ID 为空，忽略处理]"); + return false; + } + + // 2. 获取设备信息 + IotDeviceRespDTO deviceInfo = deviceService.getDeviceFromCache(message.getDeviceId()); + if (deviceInfo == null) { + log.warn("[handleDownstreamMessage][设备不存在，设备 ID：{}]", message.getDeviceId()); + return false; + } + + // 3. 构建设备标识 + String deviceKey = deviceInfo.getProductKey() + ":" + deviceInfo.getDeviceName(); + + // 4. 检查设备是否在线 + if (!connectionManager.isOnline(deviceKey)) { + log.warn("[handleDownstreamMessage][设备离线，无法发送消息，deviceKey: {}]", deviceKey); + return false; + } + + // 5. 构建主题 + String topic = buildDownstreamTopic(message, deviceInfo); + if (StrUtil.isBlank(topic)) { + log.warn("[handleDownstreamMessage][主题构建失败，设备 ID：{}，方法：{}]", + message.getDeviceId(), message.getMethod()); + return false; + } + + // 6. 检查设备是否订阅了该主题 + if (!connectionManager.isSubscribed(deviceKey, topic)) { + log.warn("[handleDownstreamMessage][设备未订阅该主题，deviceKey: {}，topic: {}]", deviceKey, topic); + return false; + } + + // 8. 编码消息 + byte[] payload = deviceMessageService.encodeDeviceMessage(message, + deviceInfo.getProductKey(), deviceInfo.getDeviceName()); + if (payload == null || payload.length == 0) { + log.warn("[handleDownstreamMessage][消息编码失败，设备 ID：{}]", message.getDeviceId()); + return false; + } + + // 9. 发送消息到设备 + return sendMessageToDevice(deviceKey, topic, payload, 1); + } catch (Exception e) { + if (message != null) { + log.error("[handleDownstreamMessage][处理下行消息异常，设备 ID：{}，错误：{}]", + message.getDeviceId(), e.getMessage(), e); + } + return false; + } + } + + /** + * 构建下行消息主题 + * + * @param message 设备消息 + * @param deviceInfo 设备信息 + * @return 主题 + */ + private String buildDownstreamTopic(IotDeviceMessage message, IotDeviceRespDTO deviceInfo) { + String method = message.getMethod(); + if (StrUtil.isBlank(method)) { + return null; + } + + // 使用工具类构建主题，支持回复消息处理 + boolean isReply = IotDeviceMessageUtils.isReplyMessage(message); + return IotMqttTopicUtils.buildTopicByMethod(method, deviceInfo.getProductKey(), + deviceInfo.getDeviceName(), isReply); + } + + /** + * 发送消息到设备 + * + * @param deviceKey 设备标识（productKey:deviceName） + * @param topic 主题 + * @param payload 消息内容 + * @param qos QoS 级别 + * @return 是否发送成功 + */ + private boolean sendMessageToDevice(String deviceKey, String topic, byte[] payload, int qos) { + // 获取设备连接 + ServerWebSocket socket = connectionManager.getConnection(deviceKey); + if (socket == null) { + log.warn("[sendMessageToDevice][设备未连接，deviceKey: {}]", deviceKey); + return false; + } + + try { + int messageId = qos > 0 ? generateMessageId() : 0; + + // 手动编码MQTT PUBLISH消息 + io.netty.buffer.ByteBuf byteBuf = io.netty.buffer.Unpooled.buffer(); + + // 固定头：消息类型(PUBLISH=3) + DUP(0) + QoS + RETAIN + int fixedHeaderByte1 = 0x30 | (qos << 1); // PUBLISH类型 + byteBuf.writeByte(fixedHeaderByte1); + + // 计算剩余长度 + int topicLength = topic.getBytes().length; + int remainingLength = 2 + topicLength + (qos > 0 ? 2 : 0) + payload.length; + + // 写入剩余长度（简化版本，假设小于128字节） + if (remainingLength < 128) { + byteBuf.writeByte(remainingLength); + } else { + // 处理大于127的情况 + int x = remainingLength; + do { + int encodedByte = x % 128; + x = x / 128; + if (x > 0) { + encodedByte = encodedByte | 128; + } + byteBuf.writeByte(encodedByte); + } while (x > 0); + } + + // 可变头：主题名称 + byteBuf.writeShort(topicLength); + byteBuf.writeBytes(topic.getBytes()); + + // 可变头：消息ID（仅QoS>0时） + if (qos > 0) { + byteBuf.writeShort(messageId); + } + + // 有效载荷 + byteBuf.writeBytes(payload); + + // 发送 + byte[] bytes = new byte[byteBuf.readableBytes()]; + byteBuf.readBytes(bytes); + byteBuf.release(); + + socket.writeBinaryMessage(Buffer.buffer(bytes)); + + log.info("[sendMessageToDevice][消息已发送到设备，deviceKey: {}，topic: {}，qos: {}，messageId: {}]", + deviceKey, topic, qos, messageId); + return true; + } catch (Exception e) { + log.error("[sendMessageToDevice][发送消息到设备失败，deviceKey: {}，topic: {}]", deviceKey, topic, e); + return false; + } + } + + /** + * 生成消息 ID + * + * @return 消息 ID + */ + private int generateMessageId() { + int id = messageIdGenerator.getAndIncrement(); + // MQTT 消息 ID 范围是 1-65535 + if (id > 65535) { + messageIdGenerator.set(1); + return 1; + } + return id; + } + +} diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/router/IotMqttWsUpstreamHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/router/IotMqttWsUpstreamHandler.java new file mode 100644 index 0000000..d9688b7 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqttws/router/IotMqttWsUpstreamHandler.java @@ -0,0 +1,774 @@ +package cn.iocoder.yudao.module.iot.gateway.protocol.mqttws.router; + +import cn.hutool.core.util.BooleanUtil; +import cn.hutool.core.util.IdUtil; +import cn.hutool.core.util.StrUtil; +import cn.hutool.extra.spring.SpringUtil; +import cn.iocoder.yudao.framework.common.pojo.CommonResult; +import cn.iocoder.yudao.module.iot.core.biz.IotDeviceCommonApi; +import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceAuthReqDTO; +import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceGetReqDTO; +import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceRespDTO; +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; +import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils; +import cn.iocoder.yudao.module.iot.gateway.protocol.mqttws.IotMqttWsUpstreamProtocol; +import cn.iocoder.yudao.module.iot.gateway.protocol.mqttws.manager.IotMqttWsConnectionManager; +import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.DecoderException; +import io.netty.handler.codec.mqtt.*; +import io.vertx.core.buffer.Buffer; +import io.vertx.core.http.ServerWebSocket; +import lombok.extern.slf4j.Slf4j; + +import java.nio.charset.StandardCharsets; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * IoT MQTT WebSocket 上行消息处理器 + *

+ * 处理来自设备的 MQTT 消息，包括： + * - CONNECT：设备连接认证 + * - PUBLISH：设备发布消息 + * - SUBSCRIBE：设备订阅主题 + * - UNSUBSCRIBE：设备取消订阅 + * - PINGREQ：心跳请求 + * - DISCONNECT：设备断开连接 + * + * @author 芋道源码 + */ +@Slf4j +public class IotMqttWsUpstreamHandler { + + private final IotMqttWsUpstreamProtocol upstreamProtocol; + + private final IotDeviceCommonApi deviceApi; + + private final IotDeviceMessageService messageService; + + private final IotMqttWsConnectionManager connectionManager; + + /** + * 存储 WebSocket 连接到 Socket ID 的映射 + * Key: WebSocket 对象 + * Value: Socket ID（UUID） + */ + private final ConcurrentHashMap socketIdMap = new ConcurrentHashMap<>(); + + /** + * 存储 Socket ID 对应的设备信息 + * Key: Socket ID（UUID） + * Value: 设备信息 + */ + private final ConcurrentHashMap socketDeviceMap = new ConcurrentHashMap<>(); + + /** + * 存储设备的消息 ID 生成器（用于 QoS > 0 的消息） + */ + private final ConcurrentHashMap deviceMessageIdMap = new ConcurrentHashMap<>(); + + /** + * MQTT 解码通道（用于解析 WebSocket 中的 MQTT 二进制消息） + */ + private final ThreadLocal decoderChannelThreadLocal = ThreadLocal + .withInitial(() -> new EmbeddedChannel(new MqttDecoder())); + + /** + * MQTT 编码通道（用于编码 MQTT 响应消息） + */ + private final ThreadLocal encoderChannelThreadLocal = ThreadLocal + .withInitial(() -> new EmbeddedChannel(MqttEncoder.INSTANCE)); + + public IotMqttWsUpstreamHandler(IotMqttWsUpstreamProtocol upstreamProtocol, + IotDeviceMessageService messageService, + IotMqttWsConnectionManager connectionManager) { + this.upstreamProtocol = upstreamProtocol; + this.deviceApi = SpringUtil.getBean(IotDeviceCommonApi.class); + this.messageService = messageService; + this.connectionManager = connectionManager; + } + + /** + * 处理 WebSocket 连接 + * + * @param socket WebSocket 连接 + */ + public void handle(ServerWebSocket socket) { + // 生成唯一的 Socket ID（因为 MQTT 使用二进制协议，textHandlerID() 会返回 null） + String socketId = IdUtil.simpleUUID(); + socketIdMap.put(socket, socketId); + + log.info("[handle][WebSocket 连接建立，socketId: {}，remoteAddress: {}]", + socketId, socket.remoteAddress()); + + // 设置二进制数据处理器 + socket.binaryMessageHandler(buffer -> { + try { + handleMqttMessage(socket, buffer); + } catch (Exception e) { + log.error("[handle][处理 MQTT 消息异常，socketId: {}]", socketId, e); + socket.close(); + } + }); + + // 设置关闭处理器 + socket.closeHandler(v -> { + socketIdMap.remove(socket); + IotDeviceRespDTO device = socketDeviceMap.remove(socketId); + if (device != null) { + String deviceKey = device.getProductKey() + ":" + device.getDeviceName(); + connectionManager.removeConnection(deviceKey); + deviceMessageIdMap.remove(deviceKey); + // 发送设备离线消息 + sendOfflineMessage(device); + log.info("[handle][WebSocket 连接关闭，deviceKey: {}，socketId: {}]", deviceKey, socketId); + } + }); + + // 设置异常处理器 + socket.exceptionHandler(e -> { + log.error("[handle][WebSocket 连接异常，socketId: {}]", socketId, e); + socketIdMap.remove(socket); + IotDeviceRespDTO device = socketDeviceMap.remove(socketId); + if (device != null) { + String deviceKey = device.getProductKey() + ":" + device.getDeviceName(); + connectionManager.removeConnection(deviceKey); + deviceMessageIdMap.remove(deviceKey); + } + socket.close(); + }); + } + + /** + * 处理 MQTT 消息 + * + * @param socket WebSocket 连接 + * @param buffer 消息缓冲区 + */ + private void handleMqttMessage(ServerWebSocket socket, Buffer buffer) { + String socketId = socketIdMap.get(socket); + ByteBuf byteBuf = Unpooled.wrappedBuffer(buffer.getBytes()); + + try { + // 使用 EmbeddedChannel 解码 MQTT 消息 + EmbeddedChannel decoderChannel = decoderChannelThreadLocal.get(); + decoderChannel.writeInbound(byteBuf.retain()); + + // 读取解码后的消息 + MqttMessage mqttMessage = decoderChannel.readInbound(); + if (mqttMessage == null) { + log.warn("[handleMqttMessage][MQTT 消息解码失败，socketId: {}]", socketId); + return; + } + + MqttMessageType messageType = mqttMessage.fixedHeader().messageType(); + log.debug("[handleMqttMessage][收到 MQTT 消息，类型: {}，socketId: {}]", messageType, socketId); + + // 根据消息类型分发处理 + switch (messageType) { + case CONNECT: + handleConnect(socket, (MqttConnectMessage) mqttMessage); + break; + case PUBLISH: + handlePublish(socket, (MqttPublishMessage) mqttMessage); + break; + case PUBACK: + handlePubAck(socket, mqttMessage); + break; + case PUBREC: + handlePubRec(socket, mqttMessage); + break; + case PUBREL: + handlePubRel(socket, mqttMessage); + break; + case PUBCOMP: + handlePubComp(socket, mqttMessage); + break; + case SUBSCRIBE: + handleSubscribe(socket, (MqttSubscribeMessage) mqttMessage); + break; + case UNSUBSCRIBE: + handleUnsubscribe(socket, (MqttUnsubscribeMessage) mqttMessage); + break; + case PINGREQ: + handlePingReq(socket); + break; + case DISCONNECT: + handleDisconnect(socket); + break; + default: + log.warn("[handleMqttMessage][不支持的消息类型: {}，socketId: {}]", messageType, socketId); + } + } catch (DecoderException e) { + log.error("[handleMqttMessage][MQTT 消息解码异常，socketId: {}]", socketId, e); + socket.close(); + } catch (Exception e) { + log.error("[handleMqttMessage][处理 MQTT 消息失败，socketId: {}]", socketId, e); + socket.close(); + } finally { + byteBuf.release(); + } + } + + /** + * 处理 CONNECT 消息（设备认证） + */ + private void handleConnect(ServerWebSocket socket, MqttConnectMessage message) { + String socketId = socketIdMap.get(socket); + try { + // 1. 解析 CONNECT 消息 + MqttConnectPayload payload = message.payload(); + String clientId = payload.clientIdentifier(); + String username = payload.userName(); + String password = payload.passwordInBytes() != null + ? new String(payload.passwordInBytes(), StandardCharsets.UTF_8) + : null; + + log.info("[handleConnect][收到 CONNECT 消息，clientId: {}，username: {}，socketId: {}]", + clientId, username, socketId); + + // 2. 设备认证 + IotDeviceRespDTO device = authenticateDevice(clientId, username, password); + if (device == null) { + log.warn("[handleConnect][设备认证失败，clientId: {}，socketId: {}]", clientId, socketId); + sendConnAck(socket, MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD); + socket.close(); + return; + } + + // 3. 保存设备信息 + socketDeviceMap.put(socketId, device); + String deviceKey = device.getProductKey() + ":" + device.getDeviceName(); + connectionManager.addConnection(deviceKey, socket, socketId); + deviceMessageIdMap.put(deviceKey, new AtomicInteger(1)); + + log.info("[handleConnect][设备认证成功，deviceId: {}，deviceKey: {}，socketId: {}]", + device.getId(), deviceKey, socketId); + + // 4. 发送 CONNACK + sendConnAck(socket, MqttConnectReturnCode.CONNECTION_ACCEPTED); + + // 5. 发送设备上线消息 + sendOnlineMessage(device); + + } catch (Exception e) { + log.error("[handleConnect][处理 CONNECT 消息失败，socketId: {}]", socketId, e); + sendConnAck(socket, MqttConnectReturnCode.CONNECTION_REFUSED_SERVER_UNAVAILABLE); + socket.close(); + } + } + + /** + * 处理 PUBLISH 消息（设备发布消息） + */ + private void handlePublish(ServerWebSocket socket, MqttPublishMessage message) { + String socketId = socketIdMap.get(socket); + IotDeviceRespDTO device = socketDeviceMap.get(socketId); + + if (device == null) { + log.warn("[handlePublish][设备未认证，socketId: {}]", socketId); + socket.close(); + return; + } + + try { + // 1. 解析 PUBLISH 消息 + MqttFixedHeader fixedHeader = message.fixedHeader(); + MqttPublishVariableHeader variableHeader = message.variableHeader(); + ByteBuf payload = message.payload(); + + String topic = variableHeader.topicName(); + int messageId = variableHeader.packetId(); + MqttQoS qos = fixedHeader.qosLevel(); + + log.debug("[handlePublish][收到 PUBLISH 消息，topic: {}，messageId: {}，QoS: {}，deviceId: {}]", + topic, messageId, qos, device.getId()); + + // 2. 读取 payload + byte[] payloadBytes = new byte[payload.readableBytes()]; + payload.readBytes(payloadBytes); + + // 3. 解码并发送消息 + IotDeviceMessage deviceMessage = messageService.decodeDeviceMessage(payloadBytes, + device.getProductKey(), device.getDeviceName()); + if (deviceMessage != null) { + deviceMessage.setServerId(upstreamProtocol.getServerId()); + messageService.sendDeviceMessage(deviceMessage, device.getProductKey(), + device.getDeviceName(), upstreamProtocol.getServerId()); + log.info("[handlePublish][设备消息已发送，method: {}，deviceId: {}]", + deviceMessage.getMethod(), device.getId()); + } + + // 4. 根据 QoS 级别发送相应的确认消息 + if (qos == MqttQoS.AT_LEAST_ONCE) { + // QoS 1：发送 PUBACK + sendPubAck(socket, messageId); + } else if (qos == MqttQoS.EXACTLY_ONCE) { + // QoS 2：发送 PUBREC + sendPubRec(socket, messageId); + } + // QoS 0 无需确认 + + } catch (Exception e) { + log.error("[handlePublish][处理 PUBLISH 消息失败，deviceId: {}]", device.getId(), e); + } + } + + /** + * 处理 PUBACK 消息（QoS 1 确认） + */ + private void handlePubAck(ServerWebSocket socket, MqttMessage message) { + String socketId = socketIdMap.get(socket); + IotDeviceRespDTO device = socketDeviceMap.get(socketId); + + if (device == null) { + log.warn("[handlePubAck][设备未认证，socketId: {}]", socketId); + socket.close(); + return; + } + + int messageId = ((MqttMessageIdVariableHeader) message.variableHeader()).messageId(); + log.debug("[handlePubAck][收到 PUBACK，messageId: {}，deviceId: {}]", messageId, device.getId()); + } + + /** + * 处理 PUBREC 消息（QoS 2 第一步确认） + */ + private void handlePubRec(ServerWebSocket socket, MqttMessage message) { + String socketId = socketIdMap.get(socket); + IotDeviceRespDTO device = socketDeviceMap.get(socketId); + + if (device == null) { + log.warn("[handlePubRec][设备未认证，socketId: {}]", socketId); + socket.close(); + return; + } + + int messageId = ((MqttMessageIdVariableHeader) message.variableHeader()).messageId(); + log.debug("[handlePubRec][收到 PUBREC，messageId: {}，deviceId: {}]", messageId, device.getId()); + + // 发送 PUBREL + sendPubRel(socket, messageId); + } + + /** + * 处理 PUBREL 消息（QoS 2 第二步） + */ + private void handlePubRel(ServerWebSocket socket, MqttMessage message) { + String socketId = socketIdMap.get(socket); + IotDeviceRespDTO device = socketDeviceMap.get(socketId); + + if (device == null) { + log.warn("[handlePubRel][设备未认证，socketId: {}]", socketId); + socket.close(); + return; + } + + int messageId = ((MqttMessageIdVariableHeader) message.variableHeader()).messageId(); + log.debug("[handlePubRel][收到 PUBREL，messageId: {}，deviceId: {}]", messageId, device.getId()); + + // 发送 PUBCOMP + sendPubComp(socket, messageId); + } + + /** + * 处理 PUBCOMP 消息（QoS 2 完成确认） + */ + private void handlePubComp(ServerWebSocket socket, MqttMessage message) { + String socketId = socketIdMap.get(socket); + IotDeviceRespDTO device = socketDeviceMap.get(socketId); + + if (device == null) { + log.warn("[handlePubComp][设备未认证，socketId: {}]", socketId); + socket.close(); + return; + } + + int messageId = ((MqttMessageIdVariableHeader) message.variableHeader()).messageId(); + log.debug("[handlePubComp][收到 PUBCOMP，messageId: {}，deviceId: {}]", messageId, device.getId()); + } + + /** + * 处理 SUBSCRIBE 消息（设备订阅主题） + */ + private void handleSubscribe(ServerWebSocket socket, MqttSubscribeMessage message) { + String socketId = socketIdMap.get(socket); + IotDeviceRespDTO device = socketDeviceMap.get(socketId); + + if (device == null) { + log.warn("[handleSubscribe][设备未认证，socketId: {}]", socketId); + socket.close(); + return; + } + + try { + // 1. 解析 SUBSCRIBE 消息 + int messageId = message.variableHeader().messageId(); + MqttSubscribePayload payload = message.payload(); + String deviceKey = device.getProductKey() + ":" + device.getDeviceName(); + + log.info("[handleSubscribe][设备订阅请求，deviceKey: {}，messageId: {}，主题数量: {}]", + deviceKey, messageId, payload.topicSubscriptions().size()); + + // 2. 构建 QoS 列表并记录订阅信息 + int[] grantedQosList = new int[payload.topicSubscriptions().size()]; + for (int i = 0; i < payload.topicSubscriptions().size(); i++) { + MqttTopicSubscription subscription = payload.topicSubscriptions().get(i); + String topic = subscription.topicFilter(); + grantedQosList[i] = subscription.qualityOfService().value(); + + // 记录订阅信息到连接管理器 + connectionManager.addSubscription(deviceKey, topic); + + log.info("[handleSubscribe][订阅主题: {}，QoS: {}，deviceKey: {}]", + topic, subscription.qualityOfService(), deviceKey); + } + + // 3. 发送 SUBACK + sendSubAck(socket, messageId, grantedQosList); + + } catch (Exception e) { + log.error("[handleSubscribe][处理 SUBSCRIBE 消息失败，deviceId: {}]", device.getId(), e); + } + } + + /** + * 处理 UNSUBSCRIBE 消息（设备取消订阅） + */ + private void handleUnsubscribe(ServerWebSocket socket, MqttUnsubscribeMessage message) { + String socketId = socketIdMap.get(socket); + IotDeviceRespDTO device = socketDeviceMap.get(socketId); + + if (device == null) { + log.warn("[handleUnsubscribe][设备未认证，socketId: {}]", socketId); + socket.close(); + return; + } + + try { + // 1. 解析 UNSUBSCRIBE 消息 + int messageId = message.variableHeader().messageId(); + MqttUnsubscribePayload payload = message.payload(); + String deviceKey = device.getProductKey() + ":" + device.getDeviceName(); + + log.info("[handleUnsubscribe][设备取消订阅，deviceKey: {}，messageId: {}，主题数量: {}]", + deviceKey, messageId, payload.topics().size()); + + // 2. 移除订阅信息 + for (String topic : payload.topics()) { + connectionManager.removeSubscription(deviceKey, topic); + log.info("[handleUnsubscribe][取消订阅主题: {}，deviceKey: {}]", topic, deviceKey); + } + + // 3. 发送 UNSUBACK + sendUnsubAck(socket, messageId); + + } catch (Exception e) { + log.error("[handleUnsubscribe][处理 UNSUBSCRIBE 消息失败，deviceId: {}]", device.getId(), e); + } + } + + /** + * 处理 PINGREQ 消息（心跳请求） + */ + private void handlePingReq(ServerWebSocket socket) { + String socketId = socketIdMap.get(socket); + IotDeviceRespDTO device = socketDeviceMap.get(socketId); + + if (device == null) { + log.warn("[handlePingReq][设备未认证，socketId: {}]", socketId); + socket.close(); + return; + } + + log.debug("[handlePingReq][收到心跳请求，deviceId: {}]", device.getId()); + + // 发送 PINGRESP + sendPingResp(socket); + } + + /** + * 处理 DISCONNECT 消息（设备断开连接） + */ + private void handleDisconnect(ServerWebSocket socket) { + String socketId = socketIdMap.get(socket); + IotDeviceRespDTO device = socketDeviceMap.remove(socketId); + + if (device != null) { + String deviceKey = device.getProductKey() + ":" + device.getDeviceName(); + connectionManager.removeConnection(deviceKey); + deviceMessageIdMap.remove(deviceKey); + sendOfflineMessage(device); + log.info("[handleDisconnect][设备主动断开连接，deviceKey: {}]", deviceKey); + } + + socket.close(); + } + + // ==================== 设备认证和状态相关方法 ==================== + + /** + * 设备认证 + */ + private IotDeviceRespDTO authenticateDevice(String clientId, String username, String password) { + try { + // 1. 参数校验 + if (StrUtil.hasEmpty(clientId, username, password)) { + log.warn("[authenticateDevice][认证参数不完整，clientId: {}，username: {}]", clientId, username); + return null; + } + + // 2. 构建认证参数并调用 API + IotDeviceAuthReqDTO authParams = new IotDeviceAuthReqDTO() + .setClientId(clientId) + .setUsername(username) + .setPassword(password); + + CommonResult authResult = deviceApi.authDevice(authParams); + if (!authResult.isSuccess() || !BooleanUtil.isTrue(authResult.getData())) { + log.warn("[authenticateDevice][设备认证失败，clientId: {}]", clientId); + return null; + } + + // 3. 获取设备信息 + IotDeviceAuthUtils.DeviceInfo deviceInfo = IotDeviceAuthUtils.parseUsername(username); + if (deviceInfo == null) { + log.warn("[authenticateDevice][用户名格式不正确，username: {}]", username); + return null; + } + + IotDeviceGetReqDTO getReqDTO = new IotDeviceGetReqDTO() + .setProductKey(deviceInfo.getProductKey()) + .setDeviceName(deviceInfo.getDeviceName()); + + CommonResult deviceResult = deviceApi.getDevice(getReqDTO); + if (!deviceResult.isSuccess() || deviceResult.getData() == null) { + log.warn("[authenticateDevice][获取设备信息失败，username: {}]", username); + return null; + } + + return deviceResult.getData(); + } catch (Exception e) { + log.error("[authenticateDevice][设备认证异常，clientId: {}]", clientId, e); + return null; + } + } + + /** + * 发送设备上线消息 + */ + private void sendOnlineMessage(IotDeviceRespDTO device) { + try { + IotDeviceMessage onlineMessage = IotDeviceMessage.buildStateUpdateOnline(); + messageService.sendDeviceMessage(onlineMessage, device.getProductKey(), + device.getDeviceName(), upstreamProtocol.getServerId()); + log.info("[sendOnlineMessage][设备上线，deviceId: {}]", device.getId()); + } catch (Exception e) { + log.error("[sendOnlineMessage][发送设备上线消息失败，deviceId: {}]", device.getId(), e); + } + } + + /** + * 发送设备离线消息 + */ + private void sendOfflineMessage(IotDeviceRespDTO device) { + try { + IotDeviceMessage offlineMessage = IotDeviceMessage.buildStateOffline(); + messageService.sendDeviceMessage(offlineMessage, device.getProductKey(), + device.getDeviceName(), upstreamProtocol.getServerId()); + log.info("[sendOfflineMessage][设备离线，deviceId: {}]", device.getId()); + } catch (Exception e) { + log.error("[sendOfflineMessage][发送设备离线消息失败，deviceId: {}]", device.getId(), e); + } + } + + // ==================== 发送响应消息的辅助方法 ==================== + + /** + * 发送 CONNACK 消息 + */ + private void sendConnAck(ServerWebSocket socket, MqttConnectReturnCode returnCode) { + try { + // 构建 CONNACK 消息 + MqttFixedHeader fixedHeader = new MqttFixedHeader( + MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0); + MqttConnAckVariableHeader variableHeader = new MqttConnAckVariableHeader(returnCode, false); + MqttConnAckMessage connAckMessage = new MqttConnAckMessage(fixedHeader, variableHeader); + + // 编码并发送 + sendMqttMessage(socket, connAckMessage); + + log.debug("[sendConnAck][发送 CONNACK 消息，returnCode: {}]", returnCode); + } catch (Exception e) { + log.error("[sendConnAck][发送 CONNACK 消息失败]", e); + } + } + + /** + * 发送 PUBACK 消息（QoS 1 确认） + */ + private void sendPubAck(ServerWebSocket socket, int messageId) { + try { + // 构建 PUBACK 消息 + MqttFixedHeader fixedHeader = new MqttFixedHeader( + MqttMessageType.PUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0); + MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(messageId); + MqttMessage pubAckMessage = new MqttMessage(fixedHeader, variableHeader); + + // 编码并发送 + sendMqttMessage(socket, pubAckMessage); + + log.debug("[sendPubAck][发送 PUBACK 消息，messageId: {}]", messageId); + } catch (Exception e) { + log.error("[sendPubAck][发送 PUBACK 消息失败，messageId: {}]", messageId, e); + } + } + + /** + * 发送 PUBREC 消息（QoS 2 第一步确认） + */ + private void sendPubRec(ServerWebSocket socket, int messageId) { + try { + // 构建 PUBREC 消息 + MqttFixedHeader fixedHeader = new MqttFixedHeader( + MqttMessageType.PUBREC, false, MqttQoS.AT_MOST_ONCE, false, 0); + MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(messageId); + MqttMessage pubRecMessage = new MqttMessage(fixedHeader, variableHeader); + + // 编码并发送 + sendMqttMessage(socket, pubRecMessage); + + log.debug("[sendPubRec][发送 PUBREC 消息，messageId: {}]", messageId); + } catch (Exception e) { + log.error("[sendPubRec][发送 PUBREC 消息失败，messageId: {}]", messageId, e); + } + } + + /** + * 发送 PUBREL 消息（QoS 2 第二步） + */ + private void sendPubRel(ServerWebSocket socket, int messageId) { + try { + // 构建 PUBREL 消息 + MqttFixedHeader fixedHeader = new MqttFixedHeader( + MqttMessageType.PUBREL, false, MqttQoS.AT_LEAST_ONCE, false, 0); + MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(messageId); + MqttMessage pubRelMessage = new MqttMessage(fixedHeader, variableHeader); + + // 编码并发送 + sendMqttMessage(socket, pubRelMessage); + + log.debug("[sendPubRel][发送 PUBREL 消息，messageId: {}]", messageId); + } catch (Exception e) { + log.error("[sendPubRel][发送 PUBREL 消息失败，messageId: {}]", messageId, e); + } + } + + /** + * 发送 PUBCOMP 消息（QoS 2 完成确认） + */ + private void sendPubComp(ServerWebSocket socket, int messageId) { + try { + // 构建 PUBCOMP 消息 + MqttFixedHeader fixedHeader = new MqttFixedHeader( + MqttMessageType.PUBCOMP, false, MqttQoS.AT_MOST_ONCE, false, 0); + MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(messageId); + MqttMessage pubCompMessage = new MqttMessage(fixedHeader, variableHeader); + + // 编码并发送 + sendMqttMessage(socket, pubCompMessage); + + log.debug("[sendPubComp][发送 PUBCOMP 消息，messageId: {}]", messageId); + } catch (Exception e) { + log.error("[sendPubComp][发送 PUBCOMP 消息失败，messageId: {}]", messageId, e); + } + } + + /** + * 发送 SUBACK 消息 + */ + private void sendSubAck(ServerWebSocket socket, int messageId, int[] grantedQosList) { + try { + // 构建 SUBACK 消息 + MqttFixedHeader fixedHeader = new MqttFixedHeader( + MqttMessageType.SUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0); + MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(messageId); + MqttSubAckPayload payload = new MqttSubAckPayload(grantedQosList); + MqttSubAckMessage subAckMessage = new MqttSubAckMessage(fixedHeader, variableHeader, payload); + + // 编码并发送 + sendMqttMessage(socket, subAckMessage); + + log.debug("[sendSubAck][发送 SUBACK 消息，messageId: {}，主题数量: {}]", messageId, grantedQosList.length); + } catch (Exception e) { + log.error("[sendSubAck][发送 SUBACK 消息失败，messageId: {}]", messageId, e); + } + } + + /** + * 发送 UNSUBACK 消息 + */ + private void sendUnsubAck(ServerWebSocket socket, int messageId) { + try { + // 构建 UNSUBACK 消息 + MqttFixedHeader fixedHeader = new MqttFixedHeader( + MqttMessageType.UNSUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0); + MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(messageId); + MqttUnsubAckMessage unsubAckMessage = new MqttUnsubAckMessage(fixedHeader, variableHeader); + + // 编码并发送 + sendMqttMessage(socket, unsubAckMessage); + + log.debug("[sendUnsubAck][发送 UNSUBACK 消息，messageId: {}]", messageId); + } catch (Exception e) { + log.error("[sendUnsubAck][发送 UNSUBACK 消息失败，messageId: {}]", messageId, e); + } + } + + /** + * 发送 PINGRESP 消息 + */ + private void sendPingResp(ServerWebSocket socket) { + try { + // 构建 PINGRESP 消息 + MqttFixedHeader fixedHeader = new MqttFixedHeader( + MqttMessageType.PINGRESP, false, MqttQoS.AT_MOST_ONCE, false, 0); + MqttMessage pingRespMessage = new MqttMessage(fixedHeader); + + // 编码并发送 + sendMqttMessage(socket, pingRespMessage); + + log.debug("[sendPingResp][发送 PINGRESP 消息]"); + } catch (Exception e) { + log.error("[sendPingResp][发送 PINGRESP 消息失败]", e); + } + } + + /** + * 发送 MQTT 消息到 WebSocket + */ + private void sendMqttMessage(ServerWebSocket socket, MqttMessage mqttMessage) { + ByteBuf byteBuf = null; + try { + // 使用 EmbeddedChannel 编码 MQTT 消息 + EmbeddedChannel encoderChannel = encoderChannelThreadLocal.get(); + encoderChannel.writeOutbound(mqttMessage); + + // 读取编码后的 ByteBuf + byteBuf = encoderChannel.readOutbound(); + if (byteBuf != null) { + byte[] bytes = new byte[byteBuf.readableBytes()]; + byteBuf.readBytes(bytes); + socket.writeBinaryMessage(Buffer.buffer(bytes)); + } + } finally { + if (byteBuf != null) { + byteBuf.release(); + } + } + } + +} diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application.yaml b/yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application.yaml index b85e84c..f633f1c 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application.yaml +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application.yaml @@ -48,7 +48,7 @@ yudao: # 针对引入的 HTTP 组件的配置 # ==================================== http: - enabled: true + enabled: false server-port: 8092 # ==================================== # 针对引入的 EMQX 组件的配置 @@ -99,11 +99,24 @@ yudao: # 针对引入的 MQTT 组件的配置 # ==================================== mqtt: - enabled: true + enabled: false port: 1883 max-message-size: 8192 connect-timeout-seconds: 60 ssl-enabled: false + # ==================================== + # 针对引入的 MQTT WebSocket 组件的配置 + # ==================================== + mqtt-ws: + enabled: true # 是否启用 MQTT WebSocket + port: 8083 # WebSocket 服务端口 + path: /mqtt # WebSocket 路径 + max-message-size: 8192 # 最大消息大小（字节） + max-frame-size: 65536 # 最大帧大小（字节） + connect-timeout-seconds: 60 # 连接超时时间（秒） + keep-alive-timeout-seconds: 300 # 保持连接超时时间（秒） + ssl-enabled: false # 是否启用 SSL（wss://） + sub-protocol: mqtt # WebSocket 子协议 --- #################### 日志相关配置 #################### @@ -123,6 +136,7 @@ logging: cn.iocoder.yudao.module.iot.gateway.protocol.emqx: DEBUG cn.iocoder.yudao.module.iot.gateway.protocol.http: DEBUG cn.iocoder.yudao.module.iot.gateway.protocol.mqtt: DEBUG + cn.iocoder.yudao.module.iot.gateway.protocol.mqttws: DEBUG # 根日志级别 root: INFO diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/test/resources/mqtt-websocket-test-client.html b/yudao-module-iot/yudao-module-iot-gateway/src/test/resources/mqtt-websocket-test-client.html new file mode 100644 index 0000000..e0853ac --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/test/resources/mqtt-websocket-test-client.html @@ -0,0 +1,888 @@ + + + + + + MQTT WebSocket 测试客户端 + + + +

+
+

🚀 MQTT WebSocket 测试客户端

+

RuoYi-Vue-Pro IoT 模块 - MQTT over WebSocket 在线测试工具

+
+ + +
+

📌 标准协议格式说明

+ +
+ +
+ +
+

📡 连接配置

+ +
+ ⚫ 未连接 +
+ +
+ + + WebSocket 地址，支持 ws:// 和 wss:// +
+ +
+ + + 格式：{productKey}.{deviceName} +
+ +
+ + + 格式：{deviceName}&{productKey} +
+ +
+ + + 设备的认证密钥（Device Secret） +
+ +
+ + + +
+ + +
+
+
0
+
发送消息数
+
+
+
0
+
接收消息数
+
+
+
0
+
错误次数
+
+
+
+ + +
+

📤 消息发布

+ +
+ + +
+ +
+ + + 标准格式：/sys/{productKey}/{deviceName}/thing/property/post +
+ +
+ + +
+ +
+ + + + Alink 协议格式：id（消息 ID）、version（协议版本）、method（方法）、params（参数） + +
+ +
+ + +
+ +

📥 主题订阅

+ +
+ + +
+ +
+ + + 标准格式：/sys/{productKey}/{deviceName}/thing/method 或使用通配符 + /sys/+/+/# +
+ +
+ + +
+ +
+ + +
+
+ + +
+

📝 日志输出

+
+
+
+
+ + + + + + + + + -- libgit2 0.21.4