oneos-v5/backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

【同步】BOOT 和 CLOUD 的功能

Browse Source
This commit is contained in:
YunaiV
2026-01-18 19:01:29 +08:00
parent 2e317b165b
commit 304b2f102a
75 changed files with 3249 additions and 155 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
yudao-module-iot/yudao-module-iot-server/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/vo/device/IotDevicePageReqVO.java
View File

@@ -31,4 +31,7 @@ public class IotDevicePageReqVO extends PageParam {
@Schema(description = "设备分组编号", example = "1024")
private Long groupId;
@Schema(description = "网关设备 ID", example = "16380")
private Long gatewayId;
}

5
yudao-module-iot/yudao-module-iot-server/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/rule/vo/data/sink/IotDataSinkPageReqVO.java
View File

@@ -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;

6
yudao-module-iot/yudao-module-iot-server/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/IotSceneRuleDO.java
View File

@@ -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;
/**
* 场景定义配置
*/

43
yudao-module-iot/yudao-module-iot-server/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/config/IotDataSinkTcpConfig.java
View File

@@ -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;
}

67
yudao-module-iot/yudao-module-iot-server/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/config/IotDataSinkWebSocketConfig.java
View File

@@ -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;
}

1
yudao-module-iot/yudao-module-iot-server/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/device/IotDeviceMapper.java
View File

@@ -31,6 +31,7 @@ public interface IotDeviceMapper extends BaseMapperX<IotDeviceDO> {
.eqIfPresent(IotDeviceDO::getDeviceType, reqVO.getDeviceType())
.likeIfPresent(IotDeviceDO::getNickname, reqVO.getNickname())
.eqIfPresent(IotDeviceDO::getState, reqVO.getStatus())
.eqIfPresent(IotDeviceDO::getGatewayId, reqVO.getGatewayId())
.apply(ObjectUtil.isNotNull(reqVO.getGroupId()), "FIND_IN_SET(" + reqVO.getGroupId() + ",group_ids) > 0")
.orderByDesc(IotDeviceDO::getId));
}

4
yudao-module-iot/yudao-module-iot-server/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/rule/IotDataRuleMapper.java
View File

@@ -35,4 +35,8 @@ public interface IotDataRuleMapper extends BaseMapperX<IotDataRuleDO> {
return selectList(IotDataRuleDO::getStatus, status);
}
default IotDataRuleDO selectByName(String name) {
return selectOne(IotDataRuleDO::getName, name);
}
}

5
yudao-module-iot/yudao-module-iot-server/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/rule/IotDataSinkMapper.java
View File

@@ -21,6 +21,7 @@ public interface IotDataSinkMapper extends BaseMapperX<IotDataSinkDO> {
return selectPage(reqVO, new LambdaQueryWrapperX<IotDataSinkDO>()
.likeIfPresent(IotDataSinkDO::getName, reqVO.getName())
.eqIfPresent(IotDataSinkDO::getStatus, reqVO.getStatus())
.eqIfPresent(IotDataSinkDO::getType, reqVO.getType())
.betweenIfPresent(IotDataSinkDO::getCreateTime, reqVO.getCreateTime())
.orderByDesc(IotDataSinkDO::getId));
}
@@ -29,4 +30,8 @@ public interface IotDataSinkMapper extends BaseMapperX<IotDataSinkDO> {
return selectList(IotDataSinkDO::getStatus, status);
}
default IotDataSinkDO selectByName(String name) {
return selectOne(IotDataSinkDO::getName, name);
}
}

4
yudao-module-iot/yudao-module-iot-server/src/main/java/cn/iocoder/yudao/module/iot/service/alert/IotAlertConfigServiceImpl.java
View File

@@ -41,7 +41,7 @@ public class IotAlertConfigServiceImpl implements IotAlertConfigService {
public Long createAlertConfig(IotAlertConfigSaveReqVO createReqVO) {
// 校验关联数据是否存在
sceneRuleService.validateSceneRuleList(createReqVO.getSceneRuleIds());
adminUserApi.validateUserList(createReqVO.getReceiveUserIds()).checkError();
adminUserApi.validateUserList(createReqVO.getReceiveUserIds());
// 插入
IotAlertConfigDO alertConfig = BeanUtils.toBean(createReqVO, IotAlertConfigDO.class);
@@ -55,7 +55,7 @@ public class IotAlertConfigServiceImpl implements IotAlertConfigService {
validateAlertConfigExists(updateReqVO.getId());
// 校验关联数据是否存在
sceneRuleService.validateSceneRuleList(updateReqVO.getSceneRuleIds());
adminUserApi.validateUserList(updateReqVO.getReceiveUserIds()).checkError();
adminUserApi.validateUserList(updateReqVO.getReceiveUserIds());
// 更新
IotAlertConfigDO updateObj = BeanUtils.toBean(updateReqVO, IotAlertConfigDO.class);

2
yudao-module-iot/yudao-module-iot-server/src/main/java/cn/iocoder/yudao/module/iot/service/device/IotDeviceServiceImpl.java
View File

@@ -382,7 +382,7 @@ public class IotDeviceServiceImpl implements IotDeviceService {
return;
}
// 2.2.2 如果存在，判断是否允许更新
if (updateSupport) {
if (!updateSupport) {
throw exception(DEVICE_KEY_EXISTS);
}
updateDevice(new IotDeviceSaveReqVO().setId(existDevice.getId())

7
yudao-module-iot/yudao-module-iot-server/src/main/java/cn/iocoder/yudao/module/iot/service/device/property/IotDevicePropertyServiceImpl.java
View File

@@ -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);
}

28
yudao-module-iot/yudao-module-iot-server/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataRuleServiceImpl.java
View File

@@ -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);
}
}
/**
* 校验数据流转规则配置
*

30
yudao-module-iot/yudao-module-iot-server/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataSinkServiceImpl.java
View File

@@ -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);

5
yudao-module-iot/yudao-module-iot-server/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotTcpDataRuleAction.java
View File

@@ -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} 实现类
* <p>
@@ -23,9 +21,6 @@ import java.time.Duration;
public class IotTcpDataRuleAction extends
IotDataRuleCacheableAction<IotDataSinkTcpConfig, IotTcpClient> {
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();

85
yudao-module-iot/yudao-module-iot-server/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotWebSocketDataRuleAction.java Normal file
View File

@@ -0,0 +1,85 @@
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;
import cn.iocoder.yudao.module.iot.service.rule.data.action.websocket.IotWebSocketClient;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
/**
* WebSocket 的 {@link IotDataRuleAction} 实现类
* <p>
* 负责将设备消息发送到外部 WebSocket 服务器
* 支持 ws:// 和 wss:// 协议，支持 JSON 和 TEXT 数据格式
* 使用连接池管理 WebSocket 连接，提高性能和资源利用率
*
* @author HUIHUI
*/
@Component
@Slf4j
public class IotWebSocketDataRuleAction extends
IotDataRuleCacheableAction<IotDataSinkWebSocketConfig, IotWebSocketClient> {
@Override
public Integer getType() {
return IotDataSinkTypeEnum.WEBSOCKET.getType();
}
@Override
protected IotWebSocketClient initProducer(IotDataSinkWebSocketConfig config) throws Exception {
// 1. 参数校验
if (StrUtil.isBlank(config.getServerUrl())) {
throw new IllegalArgumentException("WebSocket 服务器地址不能为空");
}
if (!StrUtil.startWithAny(config.getServerUrl(), "ws://", "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 客户端
// TODO @puhui999：需要加锁，保证必须连接上；
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;
}
}
}

15
yudao-module-iot/yudao-module-iot-server/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/tcp/IotTcpClient.java
View File

@@ -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;
@@ -30,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;
@@ -38,16 +40,16 @@ 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";
// TODO @puhui999：可以使用 StrUtil.defaultIfBlank 方法简化
this.dataFormat = dataFormat != null ? dataFormat : IotDataSinkTcpConfig.DEFAULT_DATA_FORMAT;
}
/**
@@ -99,9 +101,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 {

177
yudao-module-iot/yudao-module-iot-server/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/websocket/IotWebSocketClient.java Normal file
View File

@@ -0,0 +1,177 @@
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;
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 客户端
* <p>
* 负责与外部 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 : IotDataSinkWebSocketConfig.DEFAULT_CONNECT_TIMEOUT_MS;
this.sendTimeoutMs = sendTimeoutMs != null ? sendTimeoutMs : IotDataSinkWebSocketConfig.DEFAULT_SEND_TIMEOUT_MS;
this.dataFormat = dataFormat != null ? dataFormat : IotDataSinkWebSocketConfig.DEFAULT_DATA_FORMAT;
}
/**
* 连接到 WebSocket 服务器
*/
@SuppressWarnings("resource")
public void connect() throws Exception {
if (connected.get()) {
log.warn("[connect][WebSocket 客户端已经连接，无需重复连接]");
return;
}
try {
HttpClient httpClient = HttpClient.newBuilder()
.connectTimeout(Duration.ofMillis(connectTimeoutMs))
.build();
CompletableFuture<WebSocket> 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 (IotDataSinkWebSocketConfig.DEFAULT_DATA_FORMAT.equalsIgnoreCase(dataFormat)) {
messageData = JsonUtils.toJsonString(message);
} else {
messageData = message.toString();
}
// 发送消息并等待完成
CompletableFuture<WebSocket> 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() +
'}';
}
}

17
yudao-module-iot/yudao-module-iot-server/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/IotSceneRuleServiceImpl.java
View File

@@ -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,25 @@ public class IotSceneRuleServiceImpl implements IotSceneRuleService {
}
});
});
// 3. 更新最后触发时间
updateLastTriggerTime(sceneRule.getId());
});
}
/**
* 更新规则场景的最后触发时间
*
* @param id 规则场景编号
*/
private void updateLastTriggerTime(Long id) {
try {
sceneRuleMapper.updateById(new IotSceneRuleDO().setId(id).setLastTriggerTime(LocalDateTime.now()));
} catch (Exception e) {
log.error("[updateLastTriggerTime][规则场景编号({}) 更新最后触发时间异常]", id, e);
}
}
private IotSceneRuleServiceImpl getSelf() {
return SpringUtil.getBean(IotSceneRuleServiceImpl.class);
}

1
yudao-module-iot/yudao-module-iot-server/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/action/IotDeviceControlSceneRuleAction.java
View File

@@ -15,6 +15,7 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import java.util.List;
import java.util.Map;
/**
* IoT 设备属性设置的 {@link IotSceneRuleAction} 实现类

2
yudao-module-iot/yudao-module-iot-server/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/action/IotDeviceServiceInvokeSceneRuleAction.java
View File

@@ -15,7 +15,9 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* IoT 设备服务调用的 {@link IotSceneRuleAction} 实现类

11
yudao-module-iot/yudao-module-iot-server/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/matcher/trigger/IotDevicePropertyPostTriggerMatcher.java
View File

@@ -36,11 +36,12 @@ 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 中
// TODO @puhui999：可以考虑 notXXX 方法，简化代码（尽量取反）
if (!IotDeviceMessageUtils.containsIdentifier(message, trigger.getIdentifier())) {
IotSceneRuleMatcherHelper.logTriggerMatchFailure(message, trigger, "消息中不包含属性: " +
trigger.getIdentifier());
return false;
}

888
yudao-module-iot/yudao-module-iot-server/src/test/resources/mqtt-websocket-test-client.html Normal file
View File

@@ -0,0 +1,888 @@
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta content="width=device-width, initial-scale=1.0" name="viewport">
<title>MQTT WebSocket 测试客户端</title>
<style>
* {
margin: 0;
padding: 0;
box-sizing: border-box;
}
body {
font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif;
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
padding: 20px;
min-height: 100vh;
}
.container {
max-width: 1200px;
margin: 0 auto;
background: white;
border-radius: 10px;
box-shadow: 0 10px 40px rgba(0, 0, 0, 0.1);
overflow: hidden;
}
.header {
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
color: white;
padding: 30px;
text-align: center;
}
.header h1 {
font-size: 28px;
margin-bottom: 10px;
}
.header p {
opacity: 0.9;
}
.info-box {
background: #f8f9fa;
border-left: 4px solid #667eea;
padding: 15px;
margin: 20px;
border-radius: 5px;
}
.info-box h3 {
color: #667eea;
margin-bottom: 10px;
font-size: 16px;
}
.info-box ul {
margin-left: 20px;
color: #666;
font-size: 14px;
line-height: 1.6;
}
.info-box code {
background: #e9ecef;
padding: 2px 6px;
border-radius: 3px;
font-family: 'Courier New', monospace;
color: #d63384;
}
.content {
display: grid;
grid-template-columns: 1fr 1fr;
gap: 20px;
padding: 30px;
}
.panel {
background: #f8f9fa;
border-radius: 8px;
padding: 20px;
}
.panel h2 {
color: #667eea;
margin-bottom: 20px;
font-size: 20px;
border-bottom: 2px solid #667eea;
padding-bottom: 10px;
}
.form-group {
margin-bottom: 15px;
}
.form-group label {
display: block;
margin-bottom: 5px;
color: #333;
font-weight: 500;
}
.form-group input,
.form-group select,
.form-group textarea {
width: 100%;
padding: 10px;
border: 1px solid #ddd;
border-radius: 4px;
font-size: 14px;
}
.form-group input:focus,
.form-group select:focus,
.form-group textarea:focus {
outline: none;
border-color: #667eea;
}
.form-group textarea {
resize: vertical;
min-height: 80px;
font-family: monospace;
}
.btn {
padding: 10px 20px;
border: none;
border-radius: 4px;
cursor: pointer;
font-size: 14px;
font-weight: 500;
transition: all 0.3s;
}
.btn-primary {
background: #667eea;
color: white;
}
.btn-primary:hover {
background: #5568d3;
}
.btn-success {
background: #28a745;
color: white;
}
.btn-success:hover {
background: #218838;
}
.btn-danger {
background: #dc3545;
color: white;
}
.btn-danger:hover {
background: #c82333;
}
.btn-warning {
background: #ffc107;
color: #333;
}
.btn-warning:hover {
background: #e0a800;
}
.btn-group {
display: flex;
gap: 10px;
margin-top: 15px;
}
.status {
padding: 10px;
border-radius: 4px;
margin-bottom: 20px;
font-weight: 500;
}
.status.disconnected {
background: #f8d7da;
color: #721c24;
}
.status.connected {
background: #d4edda;
color: #155724;
}
.status.connecting {
background: #fff3cd;
color: #856404;
}
.log-area {
background: #1e1e1e;
color: #d4d4d4;
padding: 15px;
border-radius: 4px;
height: 400px;
overflow-y: auto;
font-family: 'Courier New', monospace;
font-size: 12px;
line-height: 1.6;
}
.log-entry {
margin-bottom: 5px;
}
.log-entry.info {
color: #4ec9b0;
}
.log-entry.success {
color: #6a9955;
}
.log-entry.error {
color: #f48771;
}
.log-entry.warning {
color: #dcdcaa;
}
.stats {
display: grid;
grid-template-columns: repeat(3, 1fr);
gap: 10px;
margin-top: 15px;
}
.stat-item {
background: white;
padding: 15px;
border-radius: 4px;
text-align: center;
}
.stat-item .value {
font-size: 24px;
font-weight: bold;
color: #667eea;
}
.stat-item .label {
font-size: 12px;
color: #666;
margin-top: 5px;
}
@media (max-width: 768px) {
.content {
grid-template-columns: 1fr;
}
}
</style>
</head>
<body>
<div class="container">
<div class="header">
<h1>🚀 MQTT WebSocket 测试客户端</h1>
<p>RuoYi-Vue-Pro IoT 模块 - MQTT over WebSocket 在线测试工具</p>
</div>
<!-- 协议格式说明 -->
<div class="info-box">
<h3>📌 标准协议格式说明</h3>
<ul>
<li><strong>Topic 格式：</strong><code>/sys/{productKey}/{deviceName}/thing/property/post</code></li>
<li><strong>Client ID 格式：</strong><code>{productKey}.{deviceName}</code> 例如：<code>zOXKLvHjUqTo7ipD.ceshi001</code>
</li>
<li><strong>Username 格式：</strong><code>{deviceName}&{productKey}</code> 例如：<code>ceshi001&zOXKLvHjUqTo7ipD</code>
</li>
<li><strong>消息格式（Alink 协议）：</strong>
<pre style="background: #e9ecef; padding: 10px; border-radius: 5px; margin-top: 5px; overflow-x: auto;">
{
"id": "消息 ID（唯一标识）",
"version": "1.0",
"method": "thing.property.post",
"params": {
"temperature": 25.5,
"humidity": 60
}
}</pre>
</li>
<li><strong>常用 Topic（下行 - 服务端推送）：</strong>
<ul style="margin-top: 5px;">
<li>属性设置：<code>/sys/{pk}/{dn}/thing/property/set</code></li>
<li>服务调用：<code>/sys/{pk}/{dn}/thing/service/invoke</code></li>
<li>配置推送：<code>/sys/{pk}/{dn}/thing/config/push</code></li>
<li>OTA 升级：<code>/sys/{pk}/{dn}/thing/ota/upgrade</code></li>
</ul>
</li>
<li><strong>常用 Topic（上行 - 设备上报）：</strong>
<ul style="margin-top: 5px;">
<li>状态更新：<code>/sys/{pk}/{dn}/thing/state/update</code></li>
<li>属性上报：<code>/sys/{pk}/{dn}/thing/property/post</code></li>
<li>事件上报：<code>/sys/{pk}/{dn}/thing/event/post</code></li>
<li>OTA 进度：<code>/sys/{pk}/{dn}/thing/ota/progress</code></li>
</ul>
</li>
</ul>
</div>
<div class="content">
<!-- 连接配置面板 -->
<div class="panel">
<h2>📡 连接配置</h2>
<div class="status disconnected" id="statusBar">
⚫ 未连接
</div>
<div class="form-group">
<label>服务器地址</label>
<input id="serverUrl" placeholder="ws://host:port/path" type="text" value="ws://localhost:8083/mqtt">
<small style="color: #666; font-size: 12px;">WebSocket 地址，支持 ws:// 和 wss://</small>
</div>
<div class="form-group">
<label>Client ID</label>
<input id="clientId" placeholder="设备客户端 ID" type="text" value="fqTn4Afs982Nak4N.jiali001">
<small style="color: #666; font-size: 12px;">格式：{productKey}.{deviceName}</small>
</div>
<div class="form-group">
<label>Username</label>
<input id="username" placeholder="用户名" type="text" value="jiali001&fqTn4Afs982Nak4N">
<small style="color: #666; font-size: 12px;">格式：{deviceName}&{productKey}</small>
</div>
<div class="form-group">
<label>Password</label>
<input id="password" placeholder="设备密钥"
type="password" value="ae10188f93febbb6b37bd57f463b2a795ae2800fab8933aef75d3c6422873f28">
<small style="color: #666; font-size: 12px;">设备的认证密钥（Device Secret）</small>
</div>
<div class="btn-group">
<button class="btn btn-success" id="connectBtn" onclick="connect()">🔌 连接</button>
<button class="btn btn-danger" disabled id="disconnectBtn" onclick="disconnect()">🔌 断开</button>
<button class="btn btn-warning" onclick="clearLogs()">🗑️ 清空日志</button>
</div>
<!-- 统计信息 -->
<div class="stats">
<div class="stat-item">
<div class="value" id="sentCount">0</div>
<div class="label">发送消息数</div>
</div>
<div class="stat-item">
<div class="value" id="receivedCount">0</div>
<div class="label">接收消息数</div>
</div>
<div class="stat-item">
<div class="value" id="errorCount">0</div>
<div class="label">错误次数</div>
</div>
</div>
</div>
<!-- 消息发布面板 -->
<div class="panel">
<h2>📤 消息发布</h2>
<div class="form-group">
<label>快捷主题选择（上行消息 - 设备 → 服务端）</label>
<select id="quickPublishTopicSelect" onchange="selectQuickPublishTopic()"
style="width: 100%; padding: 8px; border: 1px solid #ddd; border-radius: 5px;">
<option value="">-- 选择上行消息类型 --</option>
<option value="thing.state.update">设备状态更新 (thing.state.update)</option>
<option value="thing.property.post">属性上报 (thing.property.post)</option>
<option value="thing.event.post">事件上报 (thing.event.post)</option>
<option value="thing.ota.progress">OTA 升级进度 (thing.ota.progress)</option>
</select>
</div>
<div class="form-group">
<label>主题 (Topic)</label>
<input id="pubTopic" placeholder="消息主题，格式：/sys/{productKey}/{deviceName}/thing/property/post" type="text"
value="/sys/fqTn4Afs982Nak4N/jiali001/thing/property/post">
<small style="color: #666; font-size: 12px;">标准格式：/sys/{productKey}/{deviceName}/thing/property/post</small>
</div>
<div class="form-group">
<label>QoS 级别</label>
<select id="pubQos">
<option value="0">0 - 最多一次</option>
<option selected value="1">1 - 至少一次</option>
<option value="2">2 - 刚好一次</option>
</select>
</div>
<div class="form-group">
<label>消息内容 (JSON - Alink 协议格式)</label>
<textarea id="pubMessage" placeholder='Alink 协议格式消息'>{
"id": "123456789",
"version": "1.0",
"method": "thing.property.post",
"params": {
"temperature": 25.5,
"humidity": 60
}
}</textarea>
<small style="color: #666; font-size: 12px;">
Alink 协议格式：id（消息 ID）、version（协议版本）、method（方法）、params（参数）
</small>
</div>
<div class="btn-group">
<button class="btn btn-primary" onclick="publish()">📤 发布消息</button>
<button class="btn btn-success" onclick="publishSampleData()">📊 发送样例数据</button>
</div>
<h2 style="margin-top: 30px;">📥 主题订阅</h2>
<div class="form-group">
<label>快捷主题选择（下行消息 - 服务端 → 设备）</label>
<select id="quickTopicSelect" onchange="selectQuickTopic()"
style="width: 100%; padding: 8px; border: 1px solid #ddd; border-radius: 5px;">
<option value="">-- 选择下行消息类型 --</option>
<optgroup label="📥 下行消息">
<option value="thing.property.set">属性设置 (thing.property.set)</option>
<option value="thing.service.invoke">服务调用 (thing.service.invoke)</option>
<option value="thing.config.push">配置推送 (thing.config.push)</option>
<option value="thing.ota.upgrade">OTA 固件推送 (thing.ota.upgrade)</option>
</optgroup>
<optgroup label="🔄 回复主题（上行消息的回复）">
<option value="thing.property.post_reply">属性上报回复 (thing.property.post_reply)</option>
<option value="thing.event.post_reply">事件上报回复 (thing.event.post_reply)</option>
</optgroup>
<optgroup label="🔧 通配符订阅">
<option value="wildcard_all">订阅所有主题 (/sys/+/+/#)</option>
<option value="wildcard_thing">订阅所有 thing 主题 (/sys/+/+/thing/#)</option>
<option value="wildcard_reply">订阅所有回复主题 (/sys/+/+/#_reply)</option>
</optgroup>
</select>
</div>
<div class="form-group">
<label>订阅主题</label>
<input id="subTopic" placeholder="订阅主题，格式：/sys/{productKey}/{deviceName}/thing/property/set" type="text"
value="/sys/fqTn4Afs982Nak4N/jiali001/thing/property/set">
<small style="color: #666; font-size: 12px;">标准格式：/sys/{productKey}/{deviceName}/thing/method 或使用通配符
/sys/+/+/#</small>
</div>
<div class="form-group">
<label>QoS 级别</label>
<select id="subQos">
<option value="0">0 - 最多一次</option>
<option selected value="1">1 - 至少一次</option>
<option value="2">2 - 刚好一次</option>
</select>
</div>
<div class="btn-group">
<button class="btn btn-primary" onclick="subscribe()">📥 订阅</button>
<button class="btn btn-danger" onclick="unsubscribe()">❌ 取消订阅</button>
</div>
</div>
<!-- 日志面板 -->
<div class="panel" style="grid-column: 1 / -1;">
<h2>📝 日志输出</h2>
<div class="log-area" id="logArea"></div>
</div>
</div>
</div>
<!-- 使用 MQTT.js 库 -->
<script src="https://unpkg.com/mqtt/dist/mqtt.min.js"></script>
<script>
let client = null;
let sentCount = 0;
let receivedCount = 0;
let errorCount = 0;
// 添加日志
function addLog(message, type = 'info') {
const logArea = document.getElementById('logArea');
const timestamp = new Date().toLocaleTimeString();
const logEntry = document.createElement('div');
logEntry.className = `log-entry ${type}`;
logEntry.textContent = `[${timestamp}] ${message}`;
logArea.appendChild(logEntry);
logArea.scrollTop = logArea.scrollHeight;
}
// 更新状态栏
function updateStatus(status, text) {
const statusBar = document.getElementById('statusBar');
statusBar.className = `status ${status}`;
const icons = {
'disconnected': '⚫',
'connecting': '🟡',
'connected': '🟢'
};
statusBar.textContent = `${icons[status]} ${text}`;
}
// 更新统计信息
function updateStats() {
document.getElementById('sentCount').textContent = sentCount;
document.getElementById('receivedCount').textContent = receivedCount;
document.getElementById('errorCount').textContent = errorCount;
}
// 连接到服务器
function connect() {
const serverUrl = document.getElementById('serverUrl').value;
const clientId = document.getElementById('clientId').value;
const username = document.getElementById('username').value;
const password = document.getElementById('password').value;
if (!serverUrl || !clientId) {
addLog('❌ 请填写服务器地址和 Client ID', 'error');
errorCount++;
updateStats();
return;
}
updateStatus('connecting', '正在连接...');
addLog(`🔄 正在连接到 ${serverUrl}...`, 'info');
const options = {
clientId: clientId,
username: username,
password: password,
clean: true,
reconnectPeriod: 5000,
connectTimeout: 30000,
};
client = mqtt.connect(serverUrl, options);
// 连接成功
client.on('connect', () => {
updateStatus('connected', '已连接');
addLog('✅ 连接成功！', 'success');
document.getElementById('connectBtn').disabled = true;
document.getElementById('disconnectBtn').disabled = false;
});
// 接收消息
client.on('message', (topic, message) => {
receivedCount++;
updateStats();
addLog(`📥 收到消息 [${topic}]: ${message.toString()}`, 'success');
});
// 连接错误
client.on('error', (error) => {
errorCount++;
updateStats();
addLog(`❌ 连接错误: ${error.message}`, 'error');
});
// 断开连接
client.on('close', () => {
updateStatus('disconnected', '未连接');
addLog('🔌 连接已断开', 'warning');
document.getElementById('connectBtn').disabled = false;
document.getElementById('disconnectBtn').disabled = true;
});
// 离线
client.on('offline', () => {
updateStatus('disconnected', '离线');
addLog('⚠️ 客户端离线', 'warning');
});
// 重连
client.on('reconnect', () => {
updateStatus('connecting', '正在重连...');
addLog('🔄 正在重连...', 'info');
});
}
// 断开连接
function disconnect() {
if (client) {
client.end();
addLog('👋 主动断开连接', 'info');
}
}
// 发布消息
function publish() {
if (!client || !client.connected) {
addLog('❌ 请先连接到服务器', 'error');
errorCount++;
updateStats();
return;
}
const topic = document.getElementById('pubTopic').value;
const qos = parseInt(document.getElementById('pubQos').value);
const message = document.getElementById('pubMessage').value;
if (!topic || !message) {
addLog('❌ 请填写主题和消息内容', 'error');
errorCount++;
updateStats();
return;
}
// 验证 JSON 格式
try {
JSON.parse(message);
} catch (e) {
addLog('⚠️ 消息不是有效的 JSON 格式，将作为纯文本发送', 'warning');
}
client.publish(topic, message, {qos: qos}, (error) => {
if (error) {
errorCount++;
updateStats();
addLog(`❌ 发布失败: ${error.message}`, 'error');
} else {
sentCount++;
updateStats();
addLog(`📤 消息已发布 [${topic}] (QoS ${qos})`, 'success');
}
});
}
// 发送样例数据
function publishSampleData() {
// 使用 Alink 协议格式的样例数据
const sampleData = {
id: Date.now().toString(),
version: "1.0",
method: "thing.property.post",
params: {
temperature: parseFloat((20 + Math.random() * 10).toFixed(2)),
humidity: parseFloat((50 + Math.random() * 20).toFixed(2)),
pressure: parseFloat((1000 + Math.random() * 50).toFixed(2))
}
};
document.getElementById('pubMessage').value = JSON.stringify(sampleData, null, 2);
addLog('样例数据已生成（Alink 协议格式）', 'info');
publish();
}
// 获取 productKey 和 deviceName
function getDeviceInfo() {
const clientId = document.getElementById('clientId').value;
const parts = clientId.split('.');
if (parts.length !== 2) {
addLog('❌ Client ID 格式不正确（应为 {productKey}.{deviceName}），无法生成主题', 'error');
return null;
}
return {
productKey: parts[0],
deviceName: parts[1]
};
}
// 快捷主题选择（消息发布 - 上行消息）
function selectQuickPublishTopic() {
const select = document.getElementById('quickPublishTopicSelect');
const selectedValue = select.value;
console.log('[selectQuickPublishTopic] 选择的值:', selectedValue);
if (!selectedValue) {
return;
}
const deviceInfo = getDeviceInfo();
if (!deviceInfo) {
return;
}
console.log('[selectQuickPublishTopic] 设备信息:', deviceInfo);
// 构建标准主题，将枚举中的点号替换为斜杠
// 例如：thing.property.post -> /sys/{pk}/{dn}/thing/property/post
const topic = `/sys/${deviceInfo.productKey}/${deviceInfo.deviceName}/${selectedValue.replace(/\./g, '/')}`;
console.log('[selectQuickPublishTopic] 生成的主题:', topic);
const pubTopicInput = document.getElementById('pubTopic');
pubTopicInput.value = topic;
console.log('[selectQuickPublishTopic] 输入框的值已设置为:', pubTopicInput.value);
addLog(`📋 已选择发布主题: ${topic}`, 'info');
// 需要 reply 的消息类型（不在 REPLY_DISABLED 列表中）
const needsReply = [
'thing.property.post',
'thing.event.post'
];
// 如果需要 reply，自动订阅 reply 主题
if (needsReply.includes(selectedValue)) {
const replyTopic = `${topic}_reply`;
if (client && client.connected) {
// 自动订阅 reply 主题
client.subscribe(replyTopic, {qos: 1}, (err) => {
if (!err) {
addLog(`✅ 已自动订阅回复主题: ${replyTopic}`, 'success');
} else {
addLog(`❌ 自动订阅回复主题失败: ${err.message}`, 'error');
}
});
} else {
addLog(`💡 提示: 该消息需要订阅回复主题 ${replyTopic}`, 'warning');
}
}
// 重置下拉框到默认选项
select.selectedIndex = 0;
console.log('[selectQuickPublishTopic] 下拉框已重置');
}
// 快捷主题选择（主题订阅 - 下行消息）
function selectQuickTopic() {
const select = document.getElementById('quickTopicSelect');
const selectedValue = select.value;
console.log('[selectQuickTopic] 选择的值:', selectedValue);
if (!selectedValue) {
return;
}
const subTopicInput = document.getElementById('subTopic');
// 处理通配符订阅
if (selectedValue === 'wildcard_all') {
subTopicInput.value = '/sys/+/+/#';
addLog('📋 已选择订阅主题: /sys/+/+/#（订阅所有主题）', 'info');
console.log('[selectQuickTopic] 输入框的值已设置为:', subTopicInput.value);
select.selectedIndex = 0;
console.log('[selectQuickTopic] 下拉框已重置');
return;
} else if (selectedValue === 'wildcard_thing') {
subTopicInput.value = '/sys/+/+/thing/#';
addLog('📋 已选择订阅主题: /sys/+/+/thing/#（订阅所有 thing 主题）', 'info');
console.log('[selectQuickTopic] 输入框的值已设置为:', subTopicInput.value);
select.selectedIndex = 0;
console.log('[selectQuickTopic] 下拉框已重置');
return;
} else if (selectedValue === 'wildcard_reply') {
const deviceInfo = getDeviceInfo();
if (!deviceInfo) {
select.selectedIndex = 0;
return;
}
subTopicInput.value = `/sys/${deviceInfo.productKey}/${deviceInfo.deviceName}/#_reply`;
addLog(`📋 已选择订阅主题: /sys/${deviceInfo.productKey}/${deviceInfo.deviceName}/#_reply（订阅所有回复主题）`, 'info');
console.log('[selectQuickTopic] 输入框的值已设置为:', subTopicInput.value);
select.selectedIndex = 0;
console.log('[selectQuickTopic] 下拉框已重置');
return;
}
const deviceInfo = getDeviceInfo();
if (!deviceInfo) {
select.selectedIndex = 0;
return;
}
console.log('[selectQuickTopic] 设备信息:', deviceInfo);
// 构建标准主题，将枚举中的点号替换为斜杠
// 例如：thing.property.set -> /sys/{pk}/{dn}/thing/property/set
const topic = `/sys/${deviceInfo.productKey}/${deviceInfo.deviceName}/${selectedValue.replace(/\./g, '/')}`;
console.log('[selectQuickTopic] 生成的主题:', topic);
subTopicInput.value = topic;
addLog(`📋 已选择订阅主题: ${topic}`, 'info');
console.log('[selectQuickTopic] 输入框的值已设置为:', subTopicInput.value);
// 重置下拉框到默认选项
select.selectedIndex = 0;
console.log('[selectQuickTopic] 下拉框已重置');
}
// 订阅主题
function subscribe() {
if (!client || !client.connected) {
addLog('❌ 请先连接到服务器', 'error');
errorCount++;
updateStats();
return;
}
const topic = document.getElementById('subTopic').value;
const qos = parseInt(document.getElementById('subQos').value);
if (!topic) {
addLog('❌ 请填写订阅主题', 'error');
errorCount++;
updateStats();
return;
}
client.subscribe(topic, {qos: qos}, (error) => {
if (error) {
errorCount++;
updateStats();
addLog(`❌ 订阅失败: ${error.message}`, 'error');
} else {
addLog(`📥 已订阅主题 [${topic}] (QoS ${qos})`, 'success');
}
});
}
// 取消订阅
function unsubscribe() {
if (!client || !client.connected) {
addLog('❌ 请先连接到服务器', 'error');
errorCount++;
updateStats();
return;
}
const topic = document.getElementById('subTopic').value;
if (!topic) {
addLog('❌ 请填写要取消的订阅主题', 'error');
errorCount++;
updateStats();
return;
}
client.unsubscribe(topic, (error) => {
if (error) {
errorCount++;
updateStats();
addLog(`❌ 取消订阅失败: ${error.message}`, 'error');
} else {
addLog(`❌ 已取消订阅 [${topic}]`, 'info');
}
});
}
// 清空日志
function clearLogs() {
document.getElementById('logArea').innerHTML = '';
sentCount = 0;
receivedCount = 0;
errorCount = 0;
updateStats();
addLog('🗑️ 日志已清空', 'info');
}
// 页面加载完成
window.onload = function () {
addLog('👋 欢迎使用 MQTT WebSocket 测试客户端！', 'success');
addLog('📚 请配置连接参数后点击"连接"按钮', 'info');
};
// 页面关闭前断开连接
window.onbeforeunload = function () {
if (client && client.connected) {
client.end();
}
};
</script>
</body>
</html>