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 01/17] =?UTF-8?q?feat=EF=BC=9A=E3=80=90IoT=20=E7=89=A9?=
=?UTF-8?q?=E8=81=94=E7=BD=91=E3=80=91=E6=96=B0=E5=A2=9E=20MQTT=20WebSocke?=
=?UTF-8?q?t=20=E5=8D=8F=E8=AE=AE=E6=94=AF=E6=8C=81?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
yudao-dependencies/pom.xml | 2 +-
.../config/IotGatewayConfiguration.java | 42 +
.../gateway/config/IotGatewayProperties.java | 101 ++
.../mqttws/IotMqttWsDownstreamSubscriber.java | 79 ++
.../mqttws/IotMqttWsUpstreamProtocol.java | 146 +++
.../manager/IotMqttWsConnectionManager.java | 257 +++++
.../gateway/protocol/mqttws/package-info.java | 15 +
.../router/IotMqttWsDownstreamHandler.java | 221 +++++
.../router/IotMqttWsUpstreamHandler.java | 774 +++++++++++++++
.../src/main/resources/application.yaml | 18 +-
.../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 e08f834fb9..ada61302be 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 4b9c3af32c..fab4c8cc85 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 2c2000fd1f..7655a3759e 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 0000000000..302824d6df
--- /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 0000000000..6944d47dad
--- /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 0000000000..2bf61bea83
--- /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 0000000000..b9af4afe3a
--- /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 0000000000..37148de7e5
--- /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 0000000000..d9688b7974
--- /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 b85e84c170..f633f1c60b 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 0000000000..e0853ac6bf
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-gateway/src/test/resources/mqtt-websocket-test-client.html
@@ -0,0 +1,888 @@
+
+
+
+
+
+ MQTT WebSocket 测试客户端
+
+
+
+
+
+
+
+
+
📌 标准协议格式说明
+
+ - Topic 格式：
/sys/{productKey}/{deviceName}/thing/property/post
+ - Client ID 格式：
{productKey}.{deviceName} 例如：zOXKLvHjUqTo7ipD.ceshi001
+
+ - Username 格式：
{deviceName}&{productKey} 例如：ceshi001&zOXKLvHjUqTo7ipD
+
+ - 消息格式（Alink 协议）：
+
+{
+ "id": "消息 ID（唯一标识）",
+ "version": "1.0",
+ "method": "thing.property.post",
+ "params": {
+ "temperature": 25.5,
+ "humidity": 60
+ }
+}
+
+ - 常用 Topic（下行 - 服务端推送）：
+
+ - 属性设置：
/sys/{pk}/{dn}/thing/property/set
+ - 服务调用：
/sys/{pk}/{dn}/thing/service/invoke
+ - 配置推送：
/sys/{pk}/{dn}/thing/config/push
+ - OTA 升级：
/sys/{pk}/{dn}/thing/ota/upgrade
+
+
+ - 常用 Topic（上行 - 设备上报）：
+
+ - 状态更新：
/sys/{pk}/{dn}/thing/state/update
+ - 属性上报：
/sys/{pk}/{dn}/thing/property/post
+ - 事件上报：
/sys/{pk}/{dn}/thing/event/post
+ - OTA 进度：
/sys/{pk}/{dn}/thing/ota/progress
+
+
+
+
+
+
+
+
+
📡 连接配置
+
+
+ ⚫ 未连接
+
+
+
+
+
+ WebSocket 地址，支持 ws:// 和 wss://
+
+
+
+
+
+ 格式：{productKey}.{deviceName}
+
+
+
+
+
+ 格式：{deviceName}&{productKey}
+
+
+
+
+
+ 设备的认证密钥（Device Secret）
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
📤 消息发布
+
+
+
+
+
+
+
+
+
+ 标准格式：/sys/{productKey}/{deviceName}/thing/property/post
+
+
+
+
+
+
+
+
+
+
+
+ Alink 协议格式：id（消息 ID）、version（协议版本）、method（方法）、params（参数）
+
+
+
+
+
+
+
+
+
📥 主题订阅
+
+
+
+
+
+
+
+
+
+ 标准格式：/sys/{productKey}/{deviceName}/thing/method 或使用通配符
+ /sys/+/+/#
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ * 负责将设备消息发送到外部 WebSocket 服务器
+ * 支持 ws:// 和 wss:// 协议，支持 JSON 和 TEXT 数据格式
+ * 使用连接池管理 WebSocket 连接，提高性能和资源利用率
+ *
+ * @author HUIHUI
+ */
+@Component
+@Slf4j
+public class IotWebSocketDataRuleAction extends
+ IotDataRuleCacheableAction {
+
+ @Override
+ public Integer getType() {
+ return IotDataSinkTypeEnum.WEBSOCKET.getType();
+ }
+
+ @Override
+ protected IotWebSocketClient initProducer(IotDataSinkWebSocketConfig config) throws Exception {
+ // 1.1 参数校验
+ if (config.getServerUrl() == null || config.getServerUrl().trim().isEmpty()) {
+ throw new IllegalArgumentException("WebSocket 服务器地址不能为空");
+ }
+ if (!config.getServerUrl().startsWith("ws://") && !config.getServerUrl().startsWith("wss://")) {
+ throw new IllegalArgumentException("WebSocket 服务器地址必须以 ws:// 或 wss:// 开头");
+ }
+
+ // 2.1 创建 WebSocket 客户端
+ IotWebSocketClient webSocketClient = new IotWebSocketClient(
+ config.getServerUrl(),
+ config.getConnectTimeoutMs(),
+ config.getSendTimeoutMs(),
+ config.getDataFormat()
+ );
+ // 2.2 连接服务器
+ webSocketClient.connect();
+ log.info("[initProducer][WebSocket 客户端创建并连接成功，服务器: {}，数据格式: {}]",
+ config.getServerUrl(), config.getDataFormat());
+ return webSocketClient;
+ }
+
+ @Override
+ protected void closeProducer(IotWebSocketClient producer) throws Exception {
+ if (producer != null) {
+ producer.close();
+ }
+ }
+
+ @Override
+ protected void execute(IotDeviceMessage message, IotDataSinkWebSocketConfig config) throws Exception {
+ try {
+ // 1.1 获取或创建 WebSocket 客户端
+ IotWebSocketClient webSocketClient = getProducer(config);
+ // 1.2 检查连接状态，如果断开则重新连接
+ if (!webSocketClient.isConnected()) {
+ log.warn("[execute][WebSocket 连接已断开，尝试重新连接，服务器: {}]", config.getServerUrl());
+ webSocketClient.connect();
+ }
+
+ // 2.1 发送消息
+ webSocketClient.sendMessage(message);
+ // 2.2 记录发送成功日志
+ log.info("[execute][message({}) config({}) 发送成功，WebSocket 服务器: {}]",
+ message, config, config.getServerUrl());
+ } catch (Exception e) {
+ log.error("[execute][message({}) config({}) 发送失败，WebSocket 服务器: {}]",
+ message, config, config.getServerUrl(), e);
+ throw e;
+ }
+ }
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/websocket/IotWebSocketClient.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/websocket/IotWebSocketClient.java
new file mode 100644
index 0000000000..15c3cd1ae3
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/websocket/IotWebSocketClient.java
@@ -0,0 +1,175 @@
+package cn.iocoder.yudao.module.iot.service.rule.data.action.websocket;
+
+import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
+import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
+import lombok.extern.slf4j.Slf4j;
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.WebSocket;
+import java.time.Duration;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * IoT WebSocket 客户端
+ *
+ * 负责与外部 WebSocket 服务器建立连接并发送设备消息
+ * 支持 ws:// 和 wss:// 协议，支持 JSON 和 TEXT 数据格式
+ * 基于 Java 11+ 内置的 java.net.http.WebSocket 实现
+ *
+ * @author HUIHUI
+ */
+@Slf4j
+public class IotWebSocketClient implements WebSocket.Listener {
+
+ private final String serverUrl;
+ private final Integer connectTimeoutMs;
+ private final Integer sendTimeoutMs;
+ private final String dataFormat;
+
+ private WebSocket webSocket;
+ private final AtomicBoolean connected = new AtomicBoolean(false);
+ private final StringBuilder messageBuffer = new StringBuilder();
+
+ public IotWebSocketClient(String serverUrl, Integer connectTimeoutMs, Integer sendTimeoutMs, String dataFormat) {
+ this.serverUrl = serverUrl;
+ this.connectTimeoutMs = connectTimeoutMs != null ? connectTimeoutMs : 5000;
+ this.sendTimeoutMs = sendTimeoutMs != null ? sendTimeoutMs : 10000;
+ this.dataFormat = dataFormat != null ? dataFormat : "JSON";
+ }
+
+ /**
+ * 连接到 WebSocket 服务器
+ */
+ public void connect() throws Exception {
+ if (connected.get()) {
+ log.warn("[connect][WebSocket 客户端已经连接，无需重复连接]");
+ return;
+ }
+
+ try {
+ HttpClient httpClient = HttpClient.newBuilder()
+ .connectTimeout(Duration.ofMillis(connectTimeoutMs))
+ .build();
+
+ CompletableFuture future = httpClient.newWebSocketBuilder()
+ .connectTimeout(Duration.ofMillis(connectTimeoutMs))
+ .buildAsync(URI.create(serverUrl), this);
+
+ // 等待连接完成
+ webSocket = future.get(connectTimeoutMs, TimeUnit.MILLISECONDS);
+ connected.set(true);
+ log.info("[connect][WebSocket 客户端连接成功，服务器地址: {}]", serverUrl);
+ } catch (Exception e) {
+ close();
+ log.error("[connect][WebSocket 客户端连接失败，服务器地址: {}]", serverUrl, e);
+ throw e;
+ }
+ }
+
+ @Override
+ public void onOpen(WebSocket webSocket) {
+ log.debug("[onOpen][WebSocket 连接已打开]");
+ webSocket.request(1);
+ }
+
+ @Override
+ public CompletionStage onText(WebSocket webSocket, CharSequence data, boolean last) {
+ messageBuffer.append(data);
+ if (last) {
+ log.debug("[onText][收到 WebSocket 消息: {}]", messageBuffer);
+ messageBuffer.setLength(0);
+ }
+ webSocket.request(1);
+ return null;
+ }
+
+ @Override
+ public CompletionStage onClose(WebSocket webSocket, int statusCode, String reason) {
+ connected.set(false);
+ log.info("[onClose][WebSocket 连接已关闭，状态码: {}，原因: {}]", statusCode, reason);
+ return null;
+ }
+
+ @Override
+ public void onError(WebSocket webSocket, Throwable error) {
+ connected.set(false);
+ log.error("[onError][WebSocket 发生错误]", error);
+ }
+
+ /**
+ * 发送设备消息
+ *
+ * @param message 设备消息
+ * @throws Exception 发送异常
+ */
+ public void sendMessage(IotDeviceMessage message) throws Exception {
+ if (!connected.get() || webSocket == null) {
+ throw new IllegalStateException("WebSocket 客户端未连接");
+ }
+
+ try {
+ String messageData;
+ if ("JSON".equalsIgnoreCase(dataFormat)) {
+ messageData = JsonUtils.toJsonString(message);
+ } else {
+ messageData = message.toString();
+ }
+
+ // 发送消息并等待完成
+ CompletableFuture future = webSocket.sendText(messageData, true);
+ future.get(sendTimeoutMs, TimeUnit.MILLISECONDS);
+ log.debug("[sendMessage][发送消息成功，设备 ID: {}，消息长度: {}]",
+ message.getDeviceId(), messageData.length());
+ } catch (Exception e) {
+ log.error("[sendMessage][发送消息失败，设备 ID: {}]", message.getDeviceId(), e);
+ throw e;
+ }
+ }
+
+ /**
+ * 关闭连接
+ */
+ public void close() {
+ if (!connected.get() && webSocket == null) {
+ return;
+ }
+
+ try {
+ if (webSocket != null) {
+ webSocket.sendClose(WebSocket.NORMAL_CLOSURE, "客户端主动关闭")
+ .orTimeout(5, TimeUnit.SECONDS)
+ .exceptionally(e -> {
+ log.warn("[close][发送关闭帧失败]", e);
+ return null;
+ });
+ }
+ connected.set(false);
+ log.info("[close][WebSocket 客户端连接已关闭，服务器地址: {}]", serverUrl);
+ } catch (Exception e) {
+ log.error("[close][关闭 WebSocket 客户端连接异常]", e);
+ }
+ }
+
+ /**
+ * 检查连接状态
+ *
+ * @return 是否已连接
+ */
+ public boolean isConnected() {
+ return connected.get() && webSocket != null;
+ }
+
+ @Override
+ public String toString() {
+ return "IotWebSocketClient{" +
+ "serverUrl='" + serverUrl + '\'' +
+ ", dataFormat='" + dataFormat + '\'' +
+ ", connected=" + connected.get() +
+ '}';
+ }
+
+}
From 60817a6a5ba8aaa60ba63c280461dbd846cbe127 Mon Sep 17 00:00:00 2001
From: puhui999
Date: Tue, 13 Jan 2026 15:38:33 +0800
Subject: [PATCH 08/17] =?UTF-8?q?perf=EF=BC=9A=E3=80=90iot=E3=80=91IotData?=
=?UTF-8?q?SinkTcpConfig=20=E5=B8=B8=E9=87=8F=E6=9E=9A=E4=B8=BE?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
.../rule/config/IotDataSinkTcpConfig.java | 43 ++++++++++++++++---
.../rule/data/action/tcp/IotTcpClient.java | 13 +++---
2 files changed, 42 insertions(+), 14 deletions(-)
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/config/IotDataSinkTcpConfig.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/config/IotDataSinkTcpConfig.java
index 3d96f11ceb..513a987f2f 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/config/IotDataSinkTcpConfig.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/config/IotDataSinkTcpConfig.java
@@ -10,6 +10,35 @@ import lombok.Data;
@Data
public class IotDataSinkTcpConfig extends IotAbstractDataSinkConfig {
+ /**
+ * 默认连接超时时间（毫秒）
+ */
+ public static final int DEFAULT_CONNECT_TIMEOUT_MS = 5000;
+ /**
+ * 默认读取超时时间（毫秒）
+ */
+ public static final int DEFAULT_READ_TIMEOUT_MS = 10000;
+ /**
+ * 默认是否启用 SSL
+ */
+ public static final boolean DEFAULT_SSL = false;
+ /**
+ * 默认数据格式
+ */
+ public static final String DEFAULT_DATA_FORMAT = "JSON";
+ /**
+ * 默认心跳间隔时间（毫秒）
+ */
+ public static final long DEFAULT_HEARTBEAT_INTERVAL_MS = 30000L;
+ /**
+ * 默认重连间隔时间（毫秒）
+ */
+ public static final long DEFAULT_RECONNECT_INTERVAL_MS = 5000L;
+ /**
+ * 默认最大重连次数
+ */
+ public static final int DEFAULT_MAX_RECONNECT_ATTEMPTS = 3;
+
/**
* TCP 服务器地址
*/
@@ -23,17 +52,17 @@ public class IotDataSinkTcpConfig extends IotAbstractDataSinkConfig {
/**
* 连接超时时间（毫秒）
*/
- private Integer connectTimeoutMs = 5000;
+ private Integer connectTimeoutMs = DEFAULT_CONNECT_TIMEOUT_MS;
/**
* 读取超时时间（毫秒）
*/
- private Integer readTimeoutMs = 10000;
+ private Integer readTimeoutMs = DEFAULT_READ_TIMEOUT_MS;
/**
* 是否启用 SSL
*/
- private Boolean ssl = false;
+ private Boolean ssl = DEFAULT_SSL;
/**
* SSL 证书路径（当 ssl=true 时需要）
@@ -43,21 +72,21 @@ public class IotDataSinkTcpConfig extends IotAbstractDataSinkConfig {
/**
* 数据格式：JSON 或 BINARY
*/
- private String dataFormat = "JSON";
+ private String dataFormat = DEFAULT_DATA_FORMAT;
/**
* 心跳间隔时间（毫秒），0 表示不启用心跳
*/
- private Long heartbeatIntervalMs = 30000L;
+ private Long heartbeatIntervalMs = DEFAULT_HEARTBEAT_INTERVAL_MS;
/**
* 重连间隔时间（毫秒）
*/
- private Long reconnectIntervalMs = 5000L;
+ private Long reconnectIntervalMs = DEFAULT_RECONNECT_INTERVAL_MS;
/**
* 最大重连次数
*/
- private Integer maxReconnectAttempts = 3;
+ private Integer maxReconnectAttempts = DEFAULT_MAX_RECONNECT_ATTEMPTS;
}
\ No newline at end of file
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/tcp/IotTcpClient.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/tcp/IotTcpClient.java
index 1618532a4a..b417dca5a2 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/tcp/IotTcpClient.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/tcp/IotTcpClient.java
@@ -2,6 +2,7 @@ package cn.iocoder.yudao.module.iot.service.rule.data.action.tcp;
import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
+import cn.iocoder.yudao.module.iot.dal.dataobject.rule.config.IotDataSinkTcpConfig;
import lombok.extern.slf4j.Slf4j;
import javax.net.ssl.SSLSocketFactory;
@@ -38,16 +39,15 @@ public class IotTcpClient {
private BufferedReader reader;
private final AtomicBoolean connected = new AtomicBoolean(false);
- // TODO @puhui999：default 值，IotDataSinkTcpConfig.java 枚举起来哈；
public IotTcpClient(String host, Integer port, Integer connectTimeoutMs, Integer readTimeoutMs,
Boolean ssl, String sslCertPath, String dataFormat) {
this.host = host;
this.port = port;
- this.connectTimeoutMs = connectTimeoutMs != null ? connectTimeoutMs : 5000;
- this.readTimeoutMs = readTimeoutMs != null ? readTimeoutMs : 10000;
- this.ssl = ssl != null ? ssl : false;
+ this.connectTimeoutMs = connectTimeoutMs != null ? connectTimeoutMs : IotDataSinkTcpConfig.DEFAULT_CONNECT_TIMEOUT_MS;
+ this.readTimeoutMs = readTimeoutMs != null ? readTimeoutMs : IotDataSinkTcpConfig.DEFAULT_READ_TIMEOUT_MS;
+ this.ssl = ssl != null ? ssl : IotDataSinkTcpConfig.DEFAULT_SSL;
this.sslCertPath = sslCertPath;
- this.dataFormat = dataFormat != null ? dataFormat : "JSON";
+ this.dataFormat = dataFormat != null ? dataFormat : IotDataSinkTcpConfig.DEFAULT_DATA_FORMAT;
}
/**
@@ -99,9 +99,8 @@ public class IotTcpClient {
}
try {
- // TODO @puhui999：枚举值
String messageData;
- if ("JSON".equalsIgnoreCase(dataFormat)) {
+ if (IotDataSinkTcpConfig.DEFAULT_DATA_FORMAT.equalsIgnoreCase(dataFormat)) {
// JSON 格式
messageData = JsonUtils.toJsonString(message);
} else {
From 9fbced1192a453d6557085b3959acead2ee9c71c Mon Sep 17 00:00:00 2001
From: puhui999
Date: Tue, 13 Jan 2026 15:42:06 +0800
Subject: [PATCH 09/17] =?UTF-8?q?perf=EF=BC=9A=E3=80=90iot=E3=80=91IotData?=
=?UTF-8?q?SinkWebSocketConfig=20=E5=B8=B8=E9=87=8F=E6=9E=9A=E4=B8=BE?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
.../config/IotDataSinkWebSocketConfig.java | 67 ++++++++++++++++---
.../iot/enums/rule/IotDataSinkTypeEnum.java | 4 +-
.../data/action/IotTcpDataRuleAction.java | 5 --
.../action/websocket/IotWebSocketClient.java | 9 +--
4 files changed, 63 insertions(+), 22 deletions(-)
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/config/IotDataSinkWebSocketConfig.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/config/IotDataSinkWebSocketConfig.java
index f1b7e86d86..55514da7c8 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/config/IotDataSinkWebSocketConfig.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/config/IotDataSinkWebSocketConfig.java
@@ -13,6 +13,51 @@ import lombok.Data;
@Data
public class IotDataSinkWebSocketConfig extends IotAbstractDataSinkConfig {
+ /**
+ * 默认连接超时时间（毫秒）
+ */
+ public static final int DEFAULT_CONNECT_TIMEOUT_MS = 5000;
+ /**
+ * 默认发送超时时间（毫秒）
+ */
+ public static final int DEFAULT_SEND_TIMEOUT_MS = 10000;
+ /**
+ * 默认心跳间隔时间（毫秒）
+ */
+ public static final long DEFAULT_HEARTBEAT_INTERVAL_MS = 30000L;
+ /**
+ * 默认心跳消息内容
+ */
+ public static final String DEFAULT_HEARTBEAT_MESSAGE = "{\"type\":\"heartbeat\"}";
+ /**
+ * 默认是否启用 SSL 证书验证
+ */
+ public static final boolean DEFAULT_VERIFY_SSL_CERT = true;
+ /**
+ * 默认数据格式
+ */
+ public static final String DEFAULT_DATA_FORMAT = "JSON";
+ /**
+ * 默认重连间隔时间（毫秒）
+ */
+ public static final long DEFAULT_RECONNECT_INTERVAL_MS = 5000L;
+ /**
+ * 默认最大重连次数
+ */
+ public static final int DEFAULT_MAX_RECONNECT_ATTEMPTS = 3;
+ /**
+ * 默认是否启用压缩
+ */
+ public static final boolean DEFAULT_ENABLE_COMPRESSION = false;
+ /**
+ * 默认消息发送重试次数
+ */
+ public static final int DEFAULT_SEND_RETRY_COUNT = 1;
+ /**
+ * 默认消息发送重试间隔（毫秒）
+ */
+ public static final long DEFAULT_SEND_RETRY_INTERVAL_MS = 1000L;
+
/**
* WebSocket 服务器地址
* 例如：ws://localhost:8080/ws 或 wss://example.com/ws
@@ -22,22 +67,22 @@ public class IotDataSinkWebSocketConfig extends IotAbstractDataSinkConfig {
/**
* 连接超时时间（毫秒）
*/
- private Integer connectTimeoutMs = 5000;
+ private Integer connectTimeoutMs = DEFAULT_CONNECT_TIMEOUT_MS;
/**
* 发送超时时间（毫秒）
*/
- private Integer sendTimeoutMs = 10000;
+ private Integer sendTimeoutMs = DEFAULT_SEND_TIMEOUT_MS;
/**
* 心跳间隔时间（毫秒），0 表示不启用心跳
*/
- private Long heartbeatIntervalMs = 30000L;
+ private Long heartbeatIntervalMs = DEFAULT_HEARTBEAT_INTERVAL_MS;
/**
* 心跳消息内容（JSON 格式）
*/
- private String heartbeatMessage = "{\"type\":\"heartbeat\"}";
+ private String heartbeatMessage = DEFAULT_HEARTBEAT_MESSAGE;
/**
* 子协议列表（逗号分隔）
@@ -52,36 +97,36 @@ public class IotDataSinkWebSocketConfig extends IotAbstractDataSinkConfig {
/**
* 是否启用 SSL 证书验证（仅对 wss:// 生效）
*/
- private Boolean verifySslCert = true;
+ private Boolean verifySslCert = DEFAULT_VERIFY_SSL_CERT;
/**
* 数据格式：JSON 或 TEXT
*/
- private String dataFormat = "JSON";
+ private String dataFormat = DEFAULT_DATA_FORMAT;
/**
* 重连间隔时间（毫秒）
*/
- private Long reconnectIntervalMs = 5000L;
+ private Long reconnectIntervalMs = DEFAULT_RECONNECT_INTERVAL_MS;
/**
* 最大重连次数
*/
- private Integer maxReconnectAttempts = 3;
+ private Integer maxReconnectAttempts = DEFAULT_MAX_RECONNECT_ATTEMPTS;
/**
* 是否启用压缩
*/
- private Boolean enableCompression = false;
+ private Boolean enableCompression = DEFAULT_ENABLE_COMPRESSION;
/**
* 消息发送重试次数
*/
- private Integer sendRetryCount = 1;
+ private Integer sendRetryCount = DEFAULT_SEND_RETRY_COUNT;
/**
* 消息发送重试间隔（毫秒）
*/
- private Long sendRetryIntervalMs = 1000L;
+ private Long sendRetryIntervalMs = DEFAULT_SEND_RETRY_INTERVAL_MS;
}
\ No newline at end of file
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/rule/IotDataSinkTypeEnum.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/rule/IotDataSinkTypeEnum.java
index 45a557db61..440fab5f53 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/rule/IotDataSinkTypeEnum.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/rule/IotDataSinkTypeEnum.java
@@ -16,8 +16,8 @@ import java.util.Arrays;
public enum IotDataSinkTypeEnum implements ArrayValuable {
HTTP(1, "HTTP"),
- TCP(2, "TCP"), // TODO @puhui999：待实现；
- WEBSOCKET(3, "WebSocket"), // TODO @puhui999：待实现；
+ TCP(2, "TCP"),
+ WEBSOCKET(3, "WebSocket"),
MQTT(10, "MQTT"), // TODO 待实现；
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotTcpDataRuleAction.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotTcpDataRuleAction.java
index 4db6dc205a..53a3b71480 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotTcpDataRuleAction.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotTcpDataRuleAction.java
@@ -7,8 +7,6 @@ import cn.iocoder.yudao.module.iot.service.rule.data.action.tcp.IotTcpClient;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
-import java.time.Duration;
-
/**
* TCP 的 {@link IotDataRuleAction} 实现类
*
@@ -23,9 +21,6 @@ import java.time.Duration;
public class IotTcpDataRuleAction extends
IotDataRuleCacheableAction {
- private static final Duration CONNECT_TIMEOUT = Duration.ofSeconds(5);
- private static final Duration SEND_TIMEOUT = Duration.ofSeconds(10);
-
@Override
public Integer getType() {
return IotDataSinkTypeEnum.TCP.getType();
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/websocket/IotWebSocketClient.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/websocket/IotWebSocketClient.java
index 15c3cd1ae3..bed197657f 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/websocket/IotWebSocketClient.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/websocket/IotWebSocketClient.java
@@ -2,6 +2,7 @@ package cn.iocoder.yudao.module.iot.service.rule.data.action.websocket;
import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
+import cn.iocoder.yudao.module.iot.dal.dataobject.rule.config.IotDataSinkWebSocketConfig;
import lombok.extern.slf4j.Slf4j;
import java.net.URI;
@@ -36,9 +37,9 @@ public class IotWebSocketClient implements WebSocket.Listener {
public IotWebSocketClient(String serverUrl, Integer connectTimeoutMs, Integer sendTimeoutMs, String dataFormat) {
this.serverUrl = serverUrl;
- this.connectTimeoutMs = connectTimeoutMs != null ? connectTimeoutMs : 5000;
- this.sendTimeoutMs = sendTimeoutMs != null ? sendTimeoutMs : 10000;
- this.dataFormat = dataFormat != null ? dataFormat : "JSON";
+ this.connectTimeoutMs = connectTimeoutMs != null ? connectTimeoutMs : IotDataSinkWebSocketConfig.DEFAULT_CONNECT_TIMEOUT_MS;
+ this.sendTimeoutMs = sendTimeoutMs != null ? sendTimeoutMs : IotDataSinkWebSocketConfig.DEFAULT_SEND_TIMEOUT_MS;
+ this.dataFormat = dataFormat != null ? dataFormat : IotDataSinkWebSocketConfig.DEFAULT_DATA_FORMAT;
}
/**
@@ -113,7 +114,7 @@ public class IotWebSocketClient implements WebSocket.Listener {
try {
String messageData;
- if ("JSON".equalsIgnoreCase(dataFormat)) {
+ if (IotDataSinkWebSocketConfig.DEFAULT_DATA_FORMAT.equalsIgnoreCase(dataFormat)) {
messageData = JsonUtils.toJsonString(message);
} else {
messageData = message.toString();
From 908f95875df88a0e5f4e41e55c0549d760d15d62 Mon Sep 17 00:00:00 2001
From: puhui999
Date: Tue, 13 Jan 2026 15:49:19 +0800
Subject: [PATCH 10/17] =?UTF-8?q?fix=EF=BC=9A=E3=80=90iot=E3=80=91saveDevi?=
=?UTF-8?q?ceProperty=20=E4=B8=AD=EF=BC=8C=E7=B1=BB=E5=9E=8B=E8=A6=81?=
=?UTF-8?q?=E8=BD=AC=E6=8D=A2?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
.../device/property/IotDevicePropertyServiceImpl.java | 7 +++++++
1 file changed, 7 insertions(+)
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/property/IotDevicePropertyServiceImpl.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/property/IotDevicePropertyServiceImpl.java
index 8031c2a11a..4e1be3a0ca 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/property/IotDevicePropertyServiceImpl.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/property/IotDevicePropertyServiceImpl.java
@@ -1,6 +1,7 @@
package cn.iocoder.yudao.module.iot.service.device.property;
import cn.hutool.core.collection.CollUtil;
+import cn.hutool.core.convert.Convert;
import cn.hutool.core.date.LocalDateTimeUtil;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.StrUtil;
@@ -145,6 +146,12 @@ public class IotDevicePropertyServiceImpl implements IotDevicePropertyService {
IotDataSpecsDataTypeEnum.STRUCT.getDataType(), IotDataSpecsDataTypeEnum.ARRAY.getDataType())) {
// 特殊：STRUCT 和 ARRAY 类型，在 TDengine 里，有没对应数据类型，只能通过 JSON 来存储
properties.put((String) key, JsonUtils.toJsonString(value));
+ } else if (IotDataSpecsDataTypeEnum.DOUBLE.getDataType().equals(thingModel.getProperty().getDataType())) {
+ properties.put((String) key, Convert.toDouble(value));
+ } else if (IotDataSpecsDataTypeEnum.FLOAT.getDataType().equals(thingModel.getProperty().getDataType())) {
+ properties.put((String) key, Convert.toFloat(value));
+ } else if (IotDataSpecsDataTypeEnum.BOOL.getDataType().equals(thingModel.getProperty().getDataType())) {
+ properties.put((String) key, Convert.toByte(value));
} else {
properties.put((String) key, value);
}
From a7f655e1e7ba87bca575246fc543dd20247b0123 Mon Sep 17 00:00:00 2001
From: puhui999
Date: Tue, 13 Jan 2026 16:04:08 +0800
Subject: [PATCH 11/17] =?UTF-8?q?fix=EF=BC=9A=E3=80=90iot=E3=80=91?=
=?UTF-8?q?=E4=BF=AE=E5=A4=8D=E5=BD=93=E5=A4=9A=E4=B8=AA=E5=8D=8F=E8=AE=AE?=
=?UTF-8?q?=E5=90=8C=E6=97=B6=E5=90=AF=E7=94=A8=E6=97=B6=EF=BC=8C=E5=87=BA?=
=?UTF-8?q?=E7=8E=B0=20Bean=20=E5=86=B2=E7=AA=81=E7=9A=84=E9=97=AE?=
=?UTF-8?q?=E9=A2=98?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
.../config/IotGatewayConfiguration.java | 19 ++++++++++---------
1 file changed, 10 insertions(+), 9 deletions(-)
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 fab4c8cc85..3e573efdde 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
@@ -21,6 +21,7 @@ import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceService;
import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
import io.vertx.core.Vertx;
import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
@@ -59,20 +60,20 @@ public class IotGatewayConfiguration {
@Slf4j
public static class EmqxProtocolConfiguration {
- @Bean(destroyMethod = "close")
+ @Bean(name = "emqxVertx", destroyMethod = "close")
public Vertx emqxVertx() {
return Vertx.vertx();
}
@Bean
public IotEmqxAuthEventProtocol iotEmqxAuthEventProtocol(IotGatewayProperties gatewayProperties,
- Vertx emqxVertx) {
+ @Qualifier("emqxVertx") Vertx emqxVertx) {
return new IotEmqxAuthEventProtocol(gatewayProperties.getProtocol().getEmqx(), emqxVertx);
}
@Bean
public IotEmqxUpstreamProtocol iotEmqxUpstreamProtocol(IotGatewayProperties gatewayProperties,
- Vertx emqxVertx) {
+ @Qualifier("emqxVertx") Vertx emqxVertx) {
return new IotEmqxUpstreamProtocol(gatewayProperties.getProtocol().getEmqx(), emqxVertx);
}
@@ -91,7 +92,7 @@ public class IotGatewayConfiguration {
@Slf4j
public static class TcpProtocolConfiguration {
- @Bean(destroyMethod = "close")
+ @Bean(name = "tcpVertx", destroyMethod = "close")
public Vertx tcpVertx() {
return Vertx.vertx();
}
@@ -101,7 +102,7 @@ public class IotGatewayConfiguration {
IotDeviceService deviceService,
IotDeviceMessageService messageService,
IotTcpConnectionManager connectionManager,
- Vertx tcpVertx) {
+ @Qualifier("tcpVertx") Vertx tcpVertx) {
return new IotTcpUpstreamProtocol(gatewayProperties.getProtocol().getTcp(),
deviceService, messageService, connectionManager, tcpVertx);
}
@@ -126,7 +127,7 @@ public class IotGatewayConfiguration {
@Slf4j
public static class MqttProtocolConfiguration {
- @Bean(destroyMethod = "close")
+ @Bean(name = "mqttVertx", destroyMethod = "close")
public Vertx mqttVertx() {
return Vertx.vertx();
}
@@ -135,7 +136,7 @@ public class IotGatewayConfiguration {
public IotMqttUpstreamProtocol iotMqttUpstreamProtocol(IotGatewayProperties gatewayProperties,
IotDeviceMessageService messageService,
IotMqttConnectionManager connectionManager,
- Vertx mqttVertx) {
+ @Qualifier("mqttVertx") Vertx mqttVertx) {
return new IotMqttUpstreamProtocol(gatewayProperties.getProtocol().getMqtt(), messageService,
connectionManager, mqttVertx);
}
@@ -163,7 +164,7 @@ public class IotGatewayConfiguration {
@Slf4j
public static class MqttWsProtocolConfiguration {
- @Bean(destroyMethod = "close")
+ @Bean(name = "mqttWsVertx", destroyMethod = "close")
public Vertx mqttWsVertx() {
return Vertx.vertx();
}
@@ -172,7 +173,7 @@ public class IotGatewayConfiguration {
public IotMqttWsUpstreamProtocol iotMqttWsUpstreamProtocol(IotGatewayProperties gatewayProperties,
IotDeviceMessageService messageService,
IotMqttWsConnectionManager connectionManager,
- Vertx mqttWsVertx) {
+ @Qualifier("mqttWsVertx") Vertx mqttWsVertx) {
return new IotMqttWsUpstreamProtocol(gatewayProperties.getProtocol().getMqttWs(),
messageService, connectionManager, mqttWsVertx);
}
From a42202e7eb9990e62a0e4d24e4b76c68134cc2b6 Mon Sep 17 00:00:00 2001
From: puhui999
Date: Tue, 13 Jan 2026 16:31:43 +0800
Subject: [PATCH 12/17] =?UTF-8?q?fix=EF=BC=9A=E3=80=90iot=E3=80=91?=
=?UTF-8?q?=E5=B1=9E=E6=80=A7=E4=B8=8A=E6=8A=A5=E5=8F=AF=E8=83=BD=E5=90=8C?=
=?UTF-8?q?=E6=97=B6=E4=B8=8A=E6=8A=A5=E5=A4=9A=E4=B8=AA=E5=B1=9E=E6=80=A7?=
=?UTF-8?q?=EF=BC=8C=E6=89=80=E4=BB=A5=E9=9C=80=E8=A6=81=E5=88=A4=E6=96=AD?=
=?UTF-8?q?=20trigger.getIdentifier()=20=E6=98=AF=E5=90=A6=E5=9C=A8=20mess?=
=?UTF-8?q?age=20=E7=9A=84=20params=20=E4=B8=AD?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
.../IotDevicePropertyPostTriggerMatcher.java | 10 ++--
.../iot/core/util/IotDeviceMessageUtils.java | 50 +++++++++++++++++++
2 files changed, 55 insertions(+), 5 deletions(-)
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/matcher/trigger/IotDevicePropertyPostTriggerMatcher.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/matcher/trigger/IotDevicePropertyPostTriggerMatcher.java
index 27cb02a1a5..f5d461275b 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/matcher/trigger/IotDevicePropertyPostTriggerMatcher.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/matcher/trigger/IotDevicePropertyPostTriggerMatcher.java
@@ -36,11 +36,11 @@ public class IotDevicePropertyPostTriggerMatcher implements IotSceneRuleTriggerM
return false;
}
- // 1.3 检查标识符是否匹配
- String messageIdentifier = IotDeviceMessageUtils.getIdentifier(message);
- if (!IotSceneRuleMatcherHelper.isIdentifierMatched(trigger.getIdentifier(), messageIdentifier)) {
- IotSceneRuleMatcherHelper.logTriggerMatchFailure(message, trigger, "标识符不匹配，期望: " +
- trigger.getIdentifier() + ", 实际: " + messageIdentifier);
+ // 1.3 检查消息中是否包含触发器指定的属性标识符
+ // 注意：属性上报可能同时上报多个属性，所以需要判断 trigger.getIdentifier() 是否在 message 的 params 中
+ if (!IotDeviceMessageUtils.containsIdentifier(message, trigger.getIdentifier())) {
+ IotSceneRuleMatcherHelper.logTriggerMatchFailure(message, trigger, "消息中不包含属性: " +
+ trigger.getIdentifier());
return false;
}
diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/util/IotDeviceMessageUtils.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/util/IotDeviceMessageUtils.java
index 65165425c8..5c1ac26005 100644
--- a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/util/IotDeviceMessageUtils.java
+++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/util/IotDeviceMessageUtils.java
@@ -5,6 +5,7 @@ import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.system.SystemUtil;
+import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
@@ -69,6 +70,55 @@ public class IotDeviceMessageUtils {
return null;
}
+ /**
+ * 判断消息中是否包含指定的标识符
+ *
+ * 对于不同消息类型的处理：
+ * - EVENT_POST/SERVICE_INVOKE：检查 params.identifier 是否匹配
+ * - STATE_UPDATE：检查 params.state 是否匹配
+ * - PROPERTY_POST：检查 params 中是否包含该属性 key
+ *
+ * @param message 消息
+ * @param identifier 要检查的标识符
+ * @return 是否包含
+ */
+ public static boolean containsIdentifier(IotDeviceMessage message, String identifier) {
+ if (message.getParams() == null || StrUtil.isBlank(identifier)) {
+ return false;
+ }
+ // EVENT_POST / SERVICE_INVOKE / STATE_UPDATE：使用原有逻辑
+ String messageIdentifier = getIdentifier(message);
+ if (messageIdentifier != null) {
+ return identifier.equals(messageIdentifier);
+ }
+ // PROPERTY_POST：检查 params 中是否包含该属性 key
+ if (StrUtil.equals(message.getMethod(), IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod())) {
+ Map params = parseParamsToMap(message.getParams());
+ return params != null && params.containsKey(identifier);
+ }
+ return false;
+ }
+
+ /**
+ * 将 params 解析为 Map
+ *
+ * @param params 参数（可能是 Map 或 JSON 字符串）
+ * @return Map，解析失败返回 null
+ */
+ @SuppressWarnings("unchecked")
+ private static Map parseParamsToMap(Object params) {
+ if (params instanceof Map) {
+ return (Map) params;
+ }
+ if (params instanceof String) {
+ try {
+ return JsonUtils.parseObject((String) params, Map.class);
+ } catch (Exception ignored) {
+ }
+ }
+ return null;
+ }
+
/**
* 从设备消息中提取指定标识符的属性值
* - 支持多种消息格式和属性值提取策略
From 9febc2b0b087bbd97f1e9dafdd7b0ad7f2f6c391 Mon Sep 17 00:00:00 2001
From: puhui999
Date: Tue, 13 Jan 2026 16:44:24 +0800
Subject: [PATCH 13/17] =?UTF-8?q?feat=EF=BC=9A=E3=80=90iot=E3=80=91execute?=
=?UTF-8?q?SceneRuleAction=20=E6=9B=B4=E6=96=B0=E8=A7=84=E5=88=99=E5=9C=BA?=
=?UTF-8?q?=E6=99=AF=E7=9A=84=E6=9C=80=E5=90=8E=E8=A7=A6=E5=8F=91=E6=97=B6?=
=?UTF-8?q?=E9=97=B4?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
.../dal/dataobject/rule/IotSceneRuleDO.java | 6 ++++++
.../rule/scene/IotSceneRuleServiceImpl.java | 20 +++++++++++++++++++
2 files changed, 26 insertions(+)
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/IotSceneRuleDO.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/IotSceneRuleDO.java
index 94aa1eb5a3..ecf87db7a7 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/IotSceneRuleDO.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/IotSceneRuleDO.java
@@ -21,6 +21,7 @@ import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
+import java.time.LocalDateTime;
import java.util.List;
/**
@@ -56,6 +57,11 @@ public class IotSceneRuleDO extends TenantBaseDO {
*/
private Integer status;
+ /**
+ * 最后触发时间
+ */
+ private LocalDateTime lastTriggerTime;
+
/**
* 场景定义配置
*/
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/IotSceneRuleServiceImpl.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/IotSceneRuleServiceImpl.java
index 41052289a6..eb70b30480 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/IotSceneRuleServiceImpl.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/IotSceneRuleServiceImpl.java
@@ -30,6 +30,7 @@ import org.springframework.cache.annotation.Cacheable;
import org.springframework.stereotype.Service;
import org.springframework.validation.annotation.Validated;
+import java.time.LocalDateTime;
import java.util.Collection;
import java.util.List;
@@ -392,9 +393,28 @@ public class IotSceneRuleServiceImpl implements IotSceneRuleService {
}
});
});
+
+ // 3. 更新最后触发时间
+ updateLastTriggerTime(sceneRule.getId());
});
}
+ /**
+ * 更新规则场景的最后触发时间
+ *
+ * @param id 规则场景编号
+ */
+ private void updateLastTriggerTime(Long id) {
+ try {
+ IotSceneRuleDO updateObj = new IotSceneRuleDO()
+ .setId(id)
+ .setLastTriggerTime(LocalDateTime.now());
+ sceneRuleMapper.updateById(updateObj);
+ } catch (Exception e) {
+ log.error("[updateLastTriggerTime][规则场景编号({}) 更新最后触发时间异常]", id, e);
+ }
+ }
+
private IotSceneRuleServiceImpl getSelf() {
return SpringUtil.getBean(IotSceneRuleServiceImpl.class);
}
From 5086e1225f55d18491b04b8f5c8638bde4c710c7 Mon Sep 17 00:00:00 2001
From: YunaiV
Date: Tue, 13 Jan 2026 23:04:28 +0800
Subject: [PATCH 14/17] =?UTF-8?q?review=EF=BC=9A=E3=80=90iot=E3=80=91?=
=?UTF-8?q?=E2=80=9C=E4=BF=AE=E5=A4=8D=E4=BA=86=E4=B8=80=E4=BA=9B=20Iot=20?=
=?UTF-8?q?=E6=A8=A1=E5=9D=97=20TODO=20=E6=8F=90=E5=88=B0=E7=9A=84?=
=?UTF-8?q?=E9=97=AE=E9=A2=98=E2=80=9D?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
.../rule/data/action/IotWebSocketDataRuleAction.java | 8 +++++---
.../iot/service/rule/data/action/tcp/IotTcpClient.java | 2 ++
.../rule/data/action/websocket/IotWebSocketClient.java | 1 +
.../iot/service/rule/scene/IotSceneRuleServiceImpl.java | 5 +----
.../trigger/IotDevicePropertyPostTriggerMatcher.java | 1 +
5 files changed, 10 insertions(+), 7 deletions(-)
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotWebSocketDataRuleAction.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotWebSocketDataRuleAction.java
index 5e2750980c..c0445df906 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotWebSocketDataRuleAction.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotWebSocketDataRuleAction.java
@@ -1,5 +1,6 @@
package cn.iocoder.yudao.module.iot.service.rule.data.action;
+import cn.hutool.core.util.StrUtil;
import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
import cn.iocoder.yudao.module.iot.dal.dataobject.rule.config.IotDataSinkWebSocketConfig;
import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum;
@@ -28,11 +29,11 @@ public class IotWebSocketDataRuleAction extends
@Override
protected IotWebSocketClient initProducer(IotDataSinkWebSocketConfig config) throws Exception {
- // 1.1 参数校验
- if (config.getServerUrl() == null || config.getServerUrl().trim().isEmpty()) {
+ // 1. 参数校验
+ if (StrUtil.isBlank(config.getServerUrl())) {
throw new IllegalArgumentException("WebSocket 服务器地址不能为空");
}
- if (!config.getServerUrl().startsWith("ws://") && !config.getServerUrl().startsWith("wss://")) {
+ if (!StrUtil.startWithAny(config.getServerUrl(), "ws://", "wss://")) {
throw new IllegalArgumentException("WebSocket 服务器地址必须以 ws:// 或 wss:// 开头");
}
@@ -61,6 +62,7 @@ public class IotWebSocketDataRuleAction extends
protected void execute(IotDeviceMessage message, IotDataSinkWebSocketConfig config) throws Exception {
try {
// 1.1 获取或创建 WebSocket 客户端
+ // TODO @puhui999：需要加锁，保证必须连接上；
IotWebSocketClient webSocketClient = getProducer(config);
// 1.2 检查连接状态，如果断开则重新连接
if (!webSocketClient.isConnected()) {
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/tcp/IotTcpClient.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/tcp/IotTcpClient.java
index b417dca5a2..15b57b5405 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/tcp/IotTcpClient.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/tcp/IotTcpClient.java
@@ -31,6 +31,7 @@ public class IotTcpClient {
private final Integer connectTimeoutMs;
private final Integer readTimeoutMs;
private final Boolean ssl;
+ // TODO @puhui999：sslCertPath 是不是没在用？
private final String sslCertPath;
private final String dataFormat;
@@ -47,6 +48,7 @@ public class IotTcpClient {
this.readTimeoutMs = readTimeoutMs != null ? readTimeoutMs : IotDataSinkTcpConfig.DEFAULT_READ_TIMEOUT_MS;
this.ssl = ssl != null ? ssl : IotDataSinkTcpConfig.DEFAULT_SSL;
this.sslCertPath = sslCertPath;
+ // TODO @puhui999：可以使用 StrUtil.defaultIfBlank 方法简化
this.dataFormat = dataFormat != null ? dataFormat : IotDataSinkTcpConfig.DEFAULT_DATA_FORMAT;
}
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/websocket/IotWebSocketClient.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/websocket/IotWebSocketClient.java
index bed197657f..2f55d6ee74 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/websocket/IotWebSocketClient.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/websocket/IotWebSocketClient.java
@@ -45,6 +45,7 @@ public class IotWebSocketClient implements WebSocket.Listener {
/**
* 连接到 WebSocket 服务器
*/
+ @SuppressWarnings("resource")
public void connect() throws Exception {
if (connected.get()) {
log.warn("[connect][WebSocket 客户端已经连接，无需重复连接]");
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/IotSceneRuleServiceImpl.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/IotSceneRuleServiceImpl.java
index eb70b30480..f96bc9f450 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/IotSceneRuleServiceImpl.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/IotSceneRuleServiceImpl.java
@@ -406,10 +406,7 @@ public class IotSceneRuleServiceImpl implements IotSceneRuleService {
*/
private void updateLastTriggerTime(Long id) {
try {
- IotSceneRuleDO updateObj = new IotSceneRuleDO()
- .setId(id)
- .setLastTriggerTime(LocalDateTime.now());
- sceneRuleMapper.updateById(updateObj);
+ sceneRuleMapper.updateById(new IotSceneRuleDO().setId(id).setLastTriggerTime(LocalDateTime.now()));
} catch (Exception e) {
log.error("[updateLastTriggerTime][规则场景编号({}) 更新最后触发时间异常]", id, e);
}
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/matcher/trigger/IotDevicePropertyPostTriggerMatcher.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/matcher/trigger/IotDevicePropertyPostTriggerMatcher.java
index f5d461275b..d653c9c42e 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/matcher/trigger/IotDevicePropertyPostTriggerMatcher.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/matcher/trigger/IotDevicePropertyPostTriggerMatcher.java
@@ -38,6 +38,7 @@ public class IotDevicePropertyPostTriggerMatcher implements IotSceneRuleTriggerM
// 1.3 检查消息中是否包含触发器指定的属性标识符
// 注意：属性上报可能同时上报多个属性，所以需要判断 trigger.getIdentifier() 是否在 message 的 params 中
+ // TODO @puhui999：可以考虑 notXXX 方法，简化代码（尽量取反）
if (!IotDeviceMessageUtils.containsIdentifier(message, trigger.getIdentifier())) {
IotSceneRuleMatcherHelper.logTriggerMatchFailure(message, trigger, "消息中不包含属性: " +
trigger.getIdentifier());
From 832b36ac47182aa75fde0d14592fb6961e05346d Mon Sep 17 00:00:00 2001
From: haohao <1036606149@qq.com>
Date: Sat, 17 Jan 2026 11:15:59 +0800
Subject: [PATCH 15/17] =?UTF-8?q?feat=EF=BC=9A=E3=80=90iot=E3=80=91?=
=?UTF-8?q?=E6=96=B0=E5=A2=9E=E6=95=B0=E6=8D=AE=E6=B5=81=E8=BD=AC=E8=A7=84?=
=?UTF-8?q?=E5=88=99=E5=92=8C=E7=9B=AE=E7=9A=84=E5=90=8D=E7=A7=B0=E5=94=AF?=
=?UTF-8?q?=E4=B8=80=E6=80=A7=E6=A0=A1=E9=AA=8C=E5=8A=9F=E8=83=BD?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
.../iot/dal/mysql/rule/IotDataRuleMapper.java | 4 +++
.../iot/dal/mysql/rule/IotDataSinkMapper.java | 4 +++
.../module/iot/enums/ErrorCodeConstants.java | 2 ++
.../rule/data/IotDataRuleServiceImpl.java | 28 +++++++++++++++++
.../rule/data/IotDataSinkServiceImpl.java | 30 +++++++++++++++++++
5 files changed, 68 insertions(+)
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/rule/IotDataRuleMapper.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/rule/IotDataRuleMapper.java
index 7c0c17d3bc..ce2eeb04bc 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/rule/IotDataRuleMapper.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/rule/IotDataRuleMapper.java
@@ -35,4 +35,8 @@ public interface IotDataRuleMapper extends BaseMapperX {
return selectList(IotDataRuleDO::getStatus, status);
}
+ default IotDataRuleDO selectByName(String name) {
+ return selectOne(IotDataRuleDO::getName, name);
+ }
+
}
\ No newline at end of file
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/rule/IotDataSinkMapper.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/rule/IotDataSinkMapper.java
index e65001db86..b13510ecc2 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/rule/IotDataSinkMapper.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/rule/IotDataSinkMapper.java
@@ -29,4 +29,8 @@ public interface IotDataSinkMapper extends BaseMapperX {
return selectList(IotDataSinkDO::getStatus, status);
}
+ default IotDataSinkDO selectByName(String name) {
+ return selectOne(IotDataSinkDO::getName, name);
+ }
+
}
\ No newline at end of file
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/ErrorCodeConstants.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/ErrorCodeConstants.java
index d1cf60e206..025d61390e 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/ErrorCodeConstants.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/ErrorCodeConstants.java
@@ -65,10 +65,12 @@ public interface ErrorCodeConstants {
// ========== IoT 数据流转规则 1-050-010-000 ==========
ErrorCode DATA_RULE_NOT_EXISTS = new ErrorCode(1_050_010_000, "数据流转规则不存在");
+ ErrorCode DATA_RULE_NAME_EXISTS = new ErrorCode(1_050_010_001, "数据流转规则名称已存在");
// ========== IoT 数据流转目的 1-050-011-000 ==========
ErrorCode DATA_SINK_NOT_EXISTS = new ErrorCode(1_050_011_000, "数据桥梁不存在");
ErrorCode DATA_SINK_DELETE_FAIL_USED_BY_RULE = new ErrorCode(1_050_011_001, "数据流转目的正在被数据流转规则使用，无法删除");
+ ErrorCode DATA_SINK_NAME_EXISTS = new ErrorCode(1_050_011_002, "数据流转目的名称已存在");
// ========== IoT 场景联动 1-050-012-000 ==========
ErrorCode RULE_SCENE_NOT_EXISTS = new ErrorCode(1_050_012_000, "场景联动不存在");
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataRuleServiceImpl.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataRuleServiceImpl.java
index 8eafcb681a..8fc27f47df 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataRuleServiceImpl.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataRuleServiceImpl.java
@@ -32,6 +32,7 @@ import java.util.*;
import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertSet;
+import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.DATA_RULE_NAME_EXISTS;
import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.DATA_RULE_NOT_EXISTS;
/**
@@ -62,6 +63,8 @@ public class IotDataRuleServiceImpl implements IotDataRuleService {
@Override
@CacheEvict(value = RedisKeyConstants.DATA_RULE_LIST, allEntries = true)
public Long createDataRule(IotDataRuleSaveReqVO createReqVO) {
+ // 校验名称唯一
+ validateDataRuleNameUnique(null, createReqVO.getName());
// 校验数据源配置和数据目的
validateDataRuleConfig(createReqVO);
// 新增
@@ -75,6 +78,8 @@ public class IotDataRuleServiceImpl implements IotDataRuleService {
public void updateDataRule(IotDataRuleSaveReqVO updateReqVO) {
// 校验存在
validateDataRuleExists(updateReqVO.getId());
+ // 校验名称唯一
+ validateDataRuleNameUnique(updateReqVO.getId(), updateReqVO.getName());
// 校验数据源配置和数据目的
validateDataRuleConfig(updateReqVO);
@@ -98,6 +103,29 @@ public class IotDataRuleServiceImpl implements IotDataRuleService {
}
}
+ /**
+ * 校验数据流转规则名称唯一性
+ *
+ * @param id 数据流转规则编号（用于更新时排除自身）
+ * @param name 数据流转规则名称
+ */
+ private void validateDataRuleNameUnique(Long id, String name) {
+ if (StrUtil.isBlank(name)) {
+ return;
+ }
+ IotDataRuleDO dataRule = dataRuleMapper.selectByName(name);
+ if (dataRule == null) {
+ return;
+ }
+ // 如果 id 为空，说明不用比较是否为相同 id 的规则
+ if (id == null) {
+ throw exception(DATA_RULE_NAME_EXISTS);
+ }
+ if (!dataRule.getId().equals(id)) {
+ throw exception(DATA_RULE_NAME_EXISTS);
+ }
+ }
+
/**
* 校验数据流转规则配置
*
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataSinkServiceImpl.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataSinkServiceImpl.java
index 9977afba22..09e11c8226 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataSinkServiceImpl.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataSinkServiceImpl.java
@@ -1,6 +1,7 @@
package cn.iocoder.yudao.module.iot.service.rule.data;
import cn.hutool.core.collection.CollUtil;
+import cn.hutool.core.util.StrUtil;
import cn.iocoder.yudao.framework.common.pojo.PageResult;
import cn.iocoder.yudao.framework.common.util.object.BeanUtils;
import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.data.sink.IotDataSinkPageReqVO;
@@ -19,6 +20,7 @@ import java.util.List;
import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.DATA_SINK_DELETE_FAIL_USED_BY_RULE;
+import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.DATA_SINK_NAME_EXISTS;
import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.DATA_SINK_NOT_EXISTS;
/**
@@ -39,6 +41,9 @@ public class IotDataSinkServiceImpl implements IotDataSinkService {
@Override
public Long createDataSink(IotDataSinkSaveReqVO createReqVO) {
+ // 校验名称唯一
+ validateDataSinkNameUnique(null, createReqVO.getName());
+ // 新增
IotDataSinkDO dataBridge = BeanUtils.toBean(createReqVO, IotDataSinkDO.class);
dataSinkMapper.insert(dataBridge);
return dataBridge.getId();
@@ -48,6 +53,8 @@ public class IotDataSinkServiceImpl implements IotDataSinkService {
public void updateDataSink(IotDataSinkSaveReqVO updateReqVO) {
// 校验存在
validateDataBridgeExists(updateReqVO.getId());
+ // 校验名称唯一
+ validateDataSinkNameUnique(updateReqVO.getId(), updateReqVO.getName());
// 更新
IotDataSinkDO updateObj = BeanUtils.toBean(updateReqVO, IotDataSinkDO.class);
dataSinkMapper.updateById(updateObj);
@@ -71,6 +78,29 @@ public class IotDataSinkServiceImpl implements IotDataSinkService {
}
}
+ /**
+ * 校验数据流转目的名称唯一性
+ *
+ * @param id 数据流转目的编号（用于更新时排除自身）
+ * @param name 数据流转目的名称
+ */
+ private void validateDataSinkNameUnique(Long id, String name) {
+ if (StrUtil.isBlank(name)) {
+ return;
+ }
+ IotDataSinkDO dataSink = dataSinkMapper.selectByName(name);
+ if (dataSink == null) {
+ return;
+ }
+ // 如果 id 为空，说明不用比较是否为相同 id 的目的
+ if (id == null) {
+ throw exception(DATA_SINK_NAME_EXISTS);
+ }
+ if (!dataSink.getId().equals(id)) {
+ throw exception(DATA_SINK_NAME_EXISTS);
+ }
+ }
+
@Override
public IotDataSinkDO getDataSink(Long id) {
return dataSinkMapper.selectById(id);
From 3027caa1d2dbf4290f5a611ab5eff3c699af602d Mon Sep 17 00:00:00 2001
From: haohao <1036606149@qq.com>
Date: Sat, 17 Jan 2026 11:33:10 +0800
Subject: [PATCH 16/17] =?UTF-8?q?feat=EF=BC=9A=E3=80=90iot=E3=80=91?=
=?UTF-8?q?=E6=96=B0=E5=A2=9E=E6=95=B0=E6=8D=AE=E7=9B=AE=E7=9A=84=E7=B1=BB?=
=?UTF-8?q?=E5=9E=8B=E5=AD=97=E6=AE=B5=E5=8F=8A=E5=85=B6=E6=9F=A5=E8=AF=A2?=
=?UTF-8?q?=E6=9D=A1=E4=BB=B6?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
.../admin/rule/vo/data/sink/IotDataSinkPageReqVO.java | 5 +++++
.../yudao/module/iot/dal/mysql/rule/IotDataSinkMapper.java | 1 +
2 files changed, 6 insertions(+)
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/rule/vo/data/sink/IotDataSinkPageReqVO.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/rule/vo/data/sink/IotDataSinkPageReqVO.java
index 06bbecc894..8a8fcdef3d 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/rule/vo/data/sink/IotDataSinkPageReqVO.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/rule/vo/data/sink/IotDataSinkPageReqVO.java
@@ -3,6 +3,7 @@ package cn.iocoder.yudao.module.iot.controller.admin.rule.vo.data.sink;
import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum;
import cn.iocoder.yudao.framework.common.pojo.PageParam;
import cn.iocoder.yudao.framework.common.validation.InEnum;
+import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum;
import io.swagger.v3.oas.annotations.media.Schema;
import lombok.Data;
import org.springframework.format.annotation.DateTimeFormat;
@@ -22,6 +23,10 @@ public class IotDataSinkPageReqVO extends PageParam {
@InEnum(CommonStatusEnum.class)
private Integer status;
+ @Schema(description = "数据目的类型", example = "1")
+ @InEnum(IotDataSinkTypeEnum.class)
+ private Integer type;
+
@Schema(description = "创建时间")
@DateTimeFormat(pattern = FORMAT_YEAR_MONTH_DAY_HOUR_MINUTE_SECOND)
private LocalDateTime[] createTime;
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/rule/IotDataSinkMapper.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/rule/IotDataSinkMapper.java
index b13510ecc2..57e2a84595 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/rule/IotDataSinkMapper.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/rule/IotDataSinkMapper.java
@@ -21,6 +21,7 @@ public interface IotDataSinkMapper extends BaseMapperX {
return selectPage(reqVO, new LambdaQueryWrapperX()
.likeIfPresent(IotDataSinkDO::getName, reqVO.getName())
.eqIfPresent(IotDataSinkDO::getStatus, reqVO.getStatus())
+ .eqIfPresent(IotDataSinkDO::getType, reqVO.getType())
.betweenIfPresent(IotDataSinkDO::getCreateTime, reqVO.getCreateTime())
.orderByDesc(IotDataSinkDO::getId));
}
From 28a30d4b79ca6098dd3a87ad39c543b00156109f Mon Sep 17 00:00:00 2001
From: YunaiV
Date: Sun, 18 Jan 2026 18:26:31 +0800
Subject: [PATCH 17/17] =?UTF-8?q?feat=EF=BC=9A=E3=80=90iot=E3=80=91?=
=?UTF-8?q?=E9=BB=98=E8=AE=A4=E5=85=B3=E9=97=AD=20mqttws=20=E5=8D=8F?=
=?UTF-8?q?=E8=AE=AE=EF=BC=8C=3D=20=3D=20=E8=A6=81=E5=90=88=E5=B9=B6=20mas?=
=?UTF-8?q?ter=20=E4=B8=80=E4=B8=8B=EF=BC=8C=E9=81=BF=E5=85=8D=E4=B8=8D?=
=?UTF-8?q?=E4=B8=80=E5=AE=9A=E5=A4=A7=E5=AE=B6=E9=83=BD=E7=94=A8?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
.../src/main/resources/application.yaml | 4 ++--
.../src/main/resources/application-local.yaml | 15 ++++++++-------
2 files changed, 10 insertions(+), 9 deletions(-)
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 f633f1c60b..d62e573fa4 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: false
+ enabled: true
server-port: 8092
# ====================================
# 针对引入的 EMQX 组件的配置
@@ -108,7 +108,7 @@ yudao:
# 针对引入的 MQTT WebSocket 组件的配置
# ====================================
mqtt-ws:
- enabled: true # 是否启用 MQTT WebSocket
+ enabled: false # 是否启用 MQTT WebSocket
port: 8083 # WebSocket 服务端口
path: /mqtt # WebSocket 路径
max-message-size: 8192 # 最大消息大小（字节）
diff --git a/yudao-server/src/main/resources/application-local.yaml b/yudao-server/src/main/resources/application-local.yaml
index dc001c27a7..f4e9da8495 100644
--- a/yudao-server/src/main/resources/application-local.yaml
+++ b/yudao-server/src/main/resources/application-local.yaml
@@ -68,13 +68,14 @@ spring:
url: jdbc:mysql://127.0.0.1:3306/ruoyi-vue-pro?useSSL=false&serverTimezone=Asia/Shanghai&allowPublicKeyRetrieval=true&rewriteBatchedStatements=true&nullCatalogMeansCurrent=true
username: root
password: 123456
- tdengine: # IoT 数据库（需要 IoT 物联网再开启噢！）
- url: jdbc:TAOS-WS://127.0.0.1:6041/ruoyi_vue_pro?varcharAsString=true
- driver-class-name: com.taosdata.jdbc.ws.WebSocketDriver
- username: root
- password: taosdata
- druid:
- validation-query: SELECT SERVER_STATUS() # TDengine 数据源的有效性检查 SQL
+# tdengine: # IoT 数据库（需要 IoT 物联网再开启噢！）
+# lazy: true # 开启懒加载，保证启动速度
+# url: jdbc:TAOS-WS://127.0.0.1:6041/ruoyi_vue_pro?varcharAsString=true
+# driver-class-name: com.taosdata.jdbc.ws.WebSocketDriver
+# username: root
+# password: taosdata
+# druid:
+# validation-query: SELECT SERVER_STATUS() # TDengine 数据源的有效性检查 SQL
# Redis 配置。Redisson 默认的配置足够使用，一般不需要进行调优
data: