szz
/
ShiShiYiBan
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat:【IOT】新增 MQTT 协议支持及相关消息解析器，完善协议转换器功能

This commit is contained in:
haohao 2025-05-24 17:30:32 +08:00
parent fbb664026d
commit af37176d50
15 changed files with 558 additions and 109 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
yudao-module-iot/yudao-module-iot-net-components/yudao-module-iot-net-component-core/src/main/java/cn/iocoder/yudao/module/iot/net/component/core/message/IotAlinkMessage.java → yudao-module-iot/yudao-module-iot-net-components/yudao-module-iot-net-component-core/src/main/java/cn/iocoder/yudao/module/iot/net/component/core/message/IotMqttMessage.java
View File

@ -8,16 +8,15 @@ import lombok.Data;
import java.util.Map;
/**
* IoT Alink 消息模型
* IoT MQTT 消息模型
* <p>
* 基于阿里云 Alink 协议规范实现的标准消息格式
* @see <a href="https://help.aliyun.com/zh/iot/user-guide/alink-protocol-1">阿里云物联网 —— Alink 协议</a>
* 基于 MQTT 协议规范实现的标准消息格式，兼容 Alink 协议
*
* @author haohao
*/
@Data
@Builder
public class IotAlinkMessage {
public class IotMqttMessage {
/**
* 消息 ID
@ -69,11 +68,11 @@ public class IotAlinkMessage {
* @param requestId 请求 ID，为空时自动生成
* @param serviceIdentifier 服务标识符
* @param params 服务参数
* @return Alink 消息对象
* @return MQTT 消息对象
*/
public static IotAlinkMessage createServiceInvokeMessage(String requestId, String serviceIdentifier,
public static IotMqttMessage createServiceInvokeMessage(String requestId, String serviceIdentifier,
Map<String, Object> params) {
return IotAlinkMessage.builder()
return IotMqttMessage.builder()
.id(requestId != null ? requestId : generateRequestId())
.method("thing.service." + serviceIdentifier)
.params(params)
@ -85,10 +84,10 @@ public class IotAlinkMessage {
*
* @param requestId 请求 ID，为空时自动生成
* @param properties 设备属性
* @return Alink 消息对象
* @return MQTT 消息对象
*/
public static IotAlinkMessage createPropertySetMessage(String requestId, Map<String, Object> properties) {
return IotAlinkMessage.builder()
public static IotMqttMessage createPropertySetMessage(String requestId, Map<String, Object> properties) {
return IotMqttMessage.builder()
.id(requestId != null ? requestId : generateRequestId())
.method("thing.service.property.set")
.params(properties)
@ -100,13 +99,13 @@ public class IotAlinkMessage {
*
* @param requestId 请求 ID，为空时自动生成
* @param identifiers 要获取的属性标识符列表
* @return Alink 消息对象
* @return MQTT 消息对象
*/
public static IotAlinkMessage createPropertyGetMessage(String requestId, String[] identifiers) {
public static IotMqttMessage createPropertyGetMessage(String requestId, String[] identifiers) {
JSONObject params = new JSONObject();
params.set("identifiers", identifiers);
return IotAlinkMessage.builder()
return IotMqttMessage.builder()
.id(requestId != null ? requestId : generateRequestId())
.method("thing.service.property.get")
.params(params)
@ -118,10 +117,10 @@ public class IotAlinkMessage {
*
* @param requestId 请求 ID，为空时自动生成
* @param configs 设备配置
* @return Alink 消息对象
* @return MQTT 消息对象
*/
public static IotAlinkMessage createConfigSetMessage(String requestId, Map<String, Object> configs) {
return IotAlinkMessage.builder()
public static IotMqttMessage createConfigSetMessage(String requestId, Map<String, Object> configs) {
return IotMqttMessage.builder()
.id(requestId != null ? requestId : generateRequestId())
.method("thing.service.config.set")
.params(configs)
@ -133,10 +132,10 @@ public class IotAlinkMessage {
*
* @param requestId 请求 ID，为空时自动生成
* @param otaInfo OTA 升级信息
* @return Alink 消息对象
* @return MQTT 消息对象
*/
public static IotAlinkMessage createOtaUpgradeMessage(String requestId, Map<String, Object> otaInfo) {
return IotAlinkMessage.builder()
public static IotMqttMessage createOtaUpgradeMessage(String requestId, Map<String, Object> otaInfo) {
return IotMqttMessage.builder()
.id(requestId != null ? requestId : generateRequestId())
.method("thing.service.ota.upgrade")
.params(otaInfo)

6
yudao-module-iot/yudao-module-iot-net-components/yudao-module-iot-net-component-emqx/src/main/java/cn/iocoder/yudao/module/iot/net/component/emqx/downstream/IotDeviceDownstreamHandlerImpl.java
View File

@ -7,7 +7,7 @@ import cn.iocoder.yudao.framework.common.pojo.CommonResult;
import cn.iocoder.yudao.module.iot.api.device.dto.control.downstream.*;
import cn.iocoder.yudao.module.iot.net.component.core.constants.IotDeviceTopicEnum;
import cn.iocoder.yudao.module.iot.net.component.core.downstream.IotDeviceDownstreamHandler;
import cn.iocoder.yudao.module.iot.net.component.core.message.IotAlinkMessage;
import cn.iocoder.yudao.module.iot.net.component.core.message.IotMqttMessage;
import cn.iocoder.yudao.module.iot.net.component.core.util.IotNetComponentCommonUtils;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.vertx.core.buffer.Buffer;
@ -56,7 +56,7 @@ public class IotDeviceDownstreamHandlerImpl implements IotDeviceDownstreamHandle
// 构建请求消息
String requestId = StrUtil.isNotEmpty(reqDTO.getRequestId()) ? reqDTO.getRequestId()
: IotNetComponentCommonUtils.generateRequestId();
IotAlinkMessage message = IotAlinkMessage.createServiceInvokeMessage(
IotMqttMessage message = IotMqttMessage.createServiceInvokeMessage(
requestId, reqDTO.getIdentifier(), reqDTO.getParams());
// 发送消息
@ -93,7 +93,7 @@ public class IotDeviceDownstreamHandlerImpl implements IotDeviceDownstreamHandle
// 构建请求消息
String requestId = StrUtil.isNotEmpty(reqDTO.getRequestId()) ? reqDTO.getRequestId()
: IotNetComponentCommonUtils.generateRequestId();
IotAlinkMessage message = IotAlinkMessage.createPropertySetMessage(requestId, reqDTO.getProperties());
IotMqttMessage message = IotMqttMessage.createPropertySetMessage(requestId, reqDTO.getProperties());
// 发送消息
publishMessage(topic, message.toJsonObject());

254
yudao-module-iot/yudao-module-iot-protocol/README.md Normal file
View File

@ -0,0 +1,254 @@
# IoT 协议模块 (yudao-module-iot-protocol)
## 概述
本模块是物联网协议处理的核心组件，提供统一的协议解析、转换和消息处理功能。作为 `yudao-module-iot-biz` 和
`yudao-module-iot-gateway-server` 等模块的共享包，实现了协议层面的抽象和统一。
## 主要功能
### 1. 协议消息模型
- **IotMqttMessage**: 基于 MQTT 协议规范的标准消息模型（默认实现）
- **IotStandardResponse**: 统一的响应格式，支持 MQTT 和 HTTP 协议
### 2. 主题管理
- **IotTopicConstants**: 主题常量定义
- **IotTopicUtils**: MQTT 主题构建和解析工具
- **IotHttpTopicUtils**: HTTP 主题构建和解析工具
- **IotTopicParser**: 高级主题解析器，支持提取设备信息、消息类型等
### 3. 协议转换
- **IotMessageParser**: 消息解析器接口
- **IotMqttMessageParser**: MQTT 协议解析器实现（默认）
- **IotHttpMessageParser**: HTTP 协议解析器实现
- **IotProtocolConverter**: 协议转换器接口
- **DefaultIotProtocolConverter**: 默认协议转换器实现
### 4. 枚举定义
- **IotProtocolTypeEnum**: 协议类型枚举
- **IotMessageTypeEnum**: 消息类型枚举
- **IotMessageDirectionEnum**: 消息方向枚举
## 使用示例
### 1. 构建主题
#### MQTT 主题
```java
// 构建设备属性设置主题
String topic = IotTopicUtils.buildPropertySetTopic("productKey", "deviceName");
// 结果: /sys/productKey/deviceName/thing/service/property/set
// 构建事件上报主题
String eventTopic = IotTopicUtils.buildEventPostTopic("productKey", "deviceName", "temperature");
// 结果: /sys/productKey/deviceName/thing/event/temperature/post
// 获取响应主题
String replyTopic = IotTopicUtils.getReplyTopic(topic);
// 结果: /sys/productKey/deviceName/thing/service/property/set_reply
```
#### HTTP 主题
```java
// 构建属性设置路径
String propSetPath = IotHttpTopicUtils.buildPropertySetPath("productKey", "deviceName");
// 结果: /topic/sys/productKey/deviceName/thing/service/property/set
// 构建属性获取路径
String propGetPath = IotHttpTopicUtils.buildPropertyGetPath("productKey", "deviceName");
// 结果: /topic/sys/productKey/deviceName/thing/service/property/get
// 构建事件上报路径
String eventPath = IotHttpTopicUtils.buildEventPostPath("productKey", "deviceName", "alarm");
// 结果: /topic/sys/productKey/deviceName/thing/event/alarm/post
// 构建自定义主题路径
String customPath = IotHttpTopicUtils.buildCustomTopicPath("productKey", "deviceName", "user/get");
// 结果: /topic/productKey/deviceName/user/get
```
### 2. 解析主题
```java
// 解析 MQTT 主题信息
IotTopicParser.TopicInfo info = IotTopicParser.parse("/sys/pk/dn/thing/service/property/set");
System.out.
println("产品Key: "+info.getProductKey()); // pk
System.out.
println("设备名称: "+info.getDeviceName()); // dn
System.out.
println("消息类型: "+info.getMessageType()); // PROPERTY_SET
System.out.
println("消息方向: "+info.getDirection()); // DOWNSTREAM
// 解析 HTTP 主题信息
String httpPath = "/topic/sys/pk/dn/thing/service/property/set";
String actualTopic = IotHttpTopicUtils.extractActualTopic(httpPath); // /sys/pk/dn/thing/service/property/set
String productKey = IotHttpTopicUtils.parseProductKeyFromTopic(actualTopic); // pk
String deviceName = IotHttpTopicUtils.parseDeviceNameFromTopic(actualTopic); // dn
```
### 3. 创建 MQTT 消息
```java
// 创建属性设置消息
Map<String, Object> properties = new HashMap<>();
properties.
put("temperature",25.5);
IotMqttMessage message = IotMqttMessage.createPropertySetMessage("123456", properties);
// 转换为 JSON 字符串
String json = message.toJsonString();
```
### 4. HTTP 协议消息处理
#### HTTP 消息格式
```json
{
"deviceKey": "productKey/deviceName",
"messageId": "123456",
"action": "property.set",
"version": "1.0",
"data": {
"temperature": 25.5,
"humidity": 60.0
}
}
```
#### 使用 HTTP 协议解析器
```java
// 创建 HTTP 协议解析器
IotHttpMessageParser httpParser = new IotHttpMessageParser();
// 解析 HTTP 消息
String topic = "/topic/sys/productKey/deviceName/thing/service/property/set";
byte[] payload = httpMessage.getBytes(StandardCharsets.UTF_8);
IotMqttMessage message = httpParser.parse(topic, payload);
// 格式化 HTTP 响应
IotStandardResponse response = IotStandardResponse.success("123456", "property.set", data);
byte[] responseBytes = httpParser.formatResponse(response);
```
### 5. 使用协议转换器
```java
@Autowired
private IotProtocolConverter protocolConverter;
// 转换 MQTT 消息（推荐使用）
IotMqttMessage mqttMessage = protocolConverter.convertToStandardMessage(mqttTopic, mqttPayload, "mqtt");
// 转换 HTTP 消息
IotMqttMessage httpMessage = protocolConverter.convertToStandardMessage(httpTopic, httpPayload, "http");
// 创建响应
IotStandardResponse response = IotStandardResponse.success("123456", "property.set", data);
byte[] responseBytes = protocolConverter.convertFromStandardResponse(response, "mqtt");
```
### 6. 自定义协议解析器
```java
@Component
public class CustomMessageParser implements IotMessageParser {
@Override
public IotMqttMessage parse(String topic, byte[] payload) {
// 实现自定义协议解析逻辑
return null;
}
@Override
public byte[] formatResponse(IotStandardResponse response) {
// 实现自定义响应格式化逻辑
return new byte[0];
}
@Override
public boolean canHandle(String topic) {
// 判断是否能处理该主题
return topic.startsWith("/custom/");
}
}
// 注册到协议转换器
@Autowired
private DefaultIotProtocolConverter converter;
@PostConstruct
public void init() {
converter.registerParser("custom", new CustomMessageParser());
}
```
## 支持的协议类型
- **MQTT**: 标准 MQTT 协议，支持设备属性、事件、服务调用（默认协议）
- **HTTP**: HTTP 协议，支持设备通过 HTTP API 进行通信
- **MQTT_RAW**: MQTT 原始协议
- **TCP**: TCP 协议
- **UDP**: UDP 协议
- **CUSTOM**: 自定义协议
## 协议对比
| 协议类型 | 传输方式 | 消息格式 | 主题格式 | 适用场景 |
|----------|------|------|----------------------------------------------------------------------------------------------------------------------------|---------------|
| MQTT | MQTT | JSON | `/sys/{productKey}/{deviceName}/...`<br/>`/mqtt/{productKey}/{deviceName}/...`<br/>`/device/{productKey}/{deviceName}/...` | 实时性要求高的设备（推荐） |
| HTTP | HTTP | JSON | `/topic/sys/{productKey}/{deviceName}/...`<br/>`/topic/{productKey}/{deviceName}/...` | 间歇性通信的设备 |
| MQTT_RAW | MQTT | 原始 | 自定义格式 | 特殊协议设备 |
## 模块依赖
本模块是一个基础模块，依赖项最小化：
- `yudao-common`: 基础工具类
- `hutool-all`: 工具库
- `lombok`: 简化代码
- `spring-boot-starter`: Spring Boot 基础支持
## 扩展点
### 1. 自定义消息解析器
实现 `IotMessageParser` 接口，支持新的协议格式。
### 2. 自定义协议转换器
实现 `IotProtocolConverter` 接口，提供更复杂的转换逻辑。
### 3. 自定义主题格式
扩展 `IotTopicParser` 的 `parseCustomTopic` 方法，支持自定义主题格式。
## 注意事项
1. 本模块设计为无状态的工具模块，避免引入有状态的组件
2. 所有的工具类都采用静态方法，便于直接调用
3. 异常处理采用返回 null 的方式，调用方需要做好空值检查
4. 日志级别建议设置为 INFO 或 WARN，避免过多的 DEBUG 日志
5. HTTP 协议解析器使用设备标识 `deviceKey`（格式：`productKey/deviceName`）来标识设备
## 版本更新
- v1.0.0: 基础功能实现，支持 MQTT 协议和 HTTP 协议支持
- 后续版本将支持更多协议类型和高级功能

3
yudao-module-iot/yudao-module-iot-protocol/pom.xml
View File

@ -1,6 +1,7 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns="http://maven.apache.org/POM/4.0.0" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>yudao-module-iot</artifactId>
<groupId>cn.iocoder.boot</groupId>

29
yudao-module-iot/yudao-module-iot-protocol/src/main/java/cn/iocoder/yudao/module/iot/protocol/config/IotProtocolAutoConfiguration.java
View File

@ -4,8 +4,8 @@ import cn.iocoder.yudao.module.iot.protocol.convert.IotProtocolConverter;
import cn.iocoder.yudao.module.iot.protocol.convert.impl.DefaultIotProtocolConverter;
import cn.iocoder.yudao.module.iot.protocol.enums.IotProtocolTypeEnum;
import cn.iocoder.yudao.module.iot.protocol.message.IotMessageParser;
import cn.iocoder.yudao.module.iot.protocol.message.impl.IotAlinkMessageParser;
import cn.iocoder.yudao.module.iot.protocol.message.impl.IotHttpMessageParser;
import cn.iocoder.yudao.module.iot.protocol.message.impl.IotMqttMessageParser;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@ -21,20 +21,21 @@ public class IotProtocolAutoConfiguration {
/**
* Bean 名称常量
*/
public static final String IOT_ALINK_MESSAGE_PARSER_BEAN_NAME = "iotAlinkMessageParser";
public static final String IOT_MQTT_MESSAGE_PARSER_BEAN_NAME = "iotMqttMessageParser";
public static final String IOT_HTTP_MESSAGE_PARSER_BEAN_NAME = "iotHttpMessageParser";
/**
* 注册 Alink 协议消息解析器
* 注册 MQTT 协议消息解析器
*
* @return Alink 协议消息解析器
* @return MQTT 协议消息解析器
*/
@Bean
@ConditionalOnMissingBean(name = IOT_ALINK_MESSAGE_PARSER_BEAN_NAME)
public IotMessageParser iotAlinkMessageParser() {
return new IotAlinkMessageParser();
@ConditionalOnMissingBean(name = IOT_MQTT_MESSAGE_PARSER_BEAN_NAME)
public IotMessageParser iotMqttMessageParser() {
return new IotMqttMessageParser();
}
/**
* 注册 HTTP 协议消息解析器
*
@ -50,26 +51,24 @@ public class IotProtocolAutoConfiguration {
* 注册默认协议转换器
* <p>
* 如果用户没有自定义协议转换器，则使用默认实现
* 默认会注册 Alink 和 HTTP 协议解析器
* 默认会注册 MQTT 和 HTTP 协议解析器
*
* @param iotAlinkMessageParser Alink 协议解析器
* @param iotMqttMessageParser MQTT 协议解析器
* @param iotHttpMessageParser HTTP 协议解析器
* @return 默认协议转换器
*/
@Bean
@ConditionalOnMissingBean
public IotProtocolConverter iotProtocolConverter(IotMessageParser iotAlinkMessageParser,
public IotProtocolConverter iotProtocolConverter(IotMessageParser iotMqttMessageParser,
IotMessageParser iotHttpMessageParser) {
DefaultIotProtocolConverter converter = new DefaultIotProtocolConverter();
// 注册 MQTT 协议解析器（默认实现）
converter.registerParser(IotProtocolTypeEnum.MQTT.getCode(), iotMqttMessageParser);
// 注册 HTTP 协议解析器
converter.registerParser(IotProtocolTypeEnum.HTTP.getCode(), iotHttpMessageParser);
// 注意：Alink 协议解析器已经在 DefaultIotProtocolConverter 构造函数中注册
// 如果需要使用自定义的 Alink 解析器实例，可以重新注册
// converter.registerParser(IotProtocolTypeEnum.ALINK.getCode(),
// iotAlinkMessageParser);
return converter;
}
}

4
yudao-module-iot/yudao-module-iot-protocol/src/main/java/cn/iocoder/yudao/module/iot/protocol/convert/IotProtocolConverter.java
View File

@ -1,6 +1,6 @@
package cn.iocoder.yudao.module.iot.protocol.convert;
import cn.iocoder.yudao.module.iot.protocol.message.IotAlinkMessage;
import cn.iocoder.yudao.module.iot.protocol.message.IotMqttMessage;
import cn.iocoder.yudao.module.iot.protocol.message.IotStandardResponse;
/**
@ -20,7 +20,7 @@ public interface IotProtocolConverter {
* @param protocol 协议类型
* @return 标准消息对象，转换失败返回 null
*/
IotAlinkMessage convertToStandardMessage(String topic, byte[] payload, String protocol);
IotMqttMessage convertToStandardMessage(String topic, byte[] payload, String protocol);
/**
* 将标准响应转换为字节数组

15
yudao-module-iot/yudao-module-iot-protocol/src/main/java/cn/iocoder/yudao/module/iot/protocol/convert/impl/DefaultIotProtocolConverter.java
View File

@ -3,10 +3,10 @@ package cn.iocoder.yudao.module.iot.protocol.convert.impl;
import cn.iocoder.yudao.module.iot.protocol.constants.IotLogConstants;
import cn.iocoder.yudao.module.iot.protocol.convert.IotProtocolConverter;
import cn.iocoder.yudao.module.iot.protocol.enums.IotProtocolTypeEnum;
import cn.iocoder.yudao.module.iot.protocol.message.IotAlinkMessage;
import cn.iocoder.yudao.module.iot.protocol.message.IotMessageParser;
import cn.iocoder.yudao.module.iot.protocol.message.IotMqttMessage;
import cn.iocoder.yudao.module.iot.protocol.message.IotStandardResponse;
import cn.iocoder.yudao.module.iot.protocol.message.impl.IotAlinkMessageParser;
import cn.iocoder.yudao.module.iot.protocol.message.impl.IotMqttMessageParser;
import lombok.extern.slf4j.Slf4j;
import java.util.HashMap;
@ -33,8 +33,9 @@ public class DefaultIotProtocolConverter implements IotProtocolConverter {
* 构造函数，初始化默认支持的协议
*/
public DefaultIotProtocolConverter() {
// 注册 Alink 协议解析器
registerParser(IotProtocolTypeEnum.ALINK.getCode(), new IotAlinkMessageParser());
// 注册 MQTT 协议解析器作为默认实现
IotMqttMessageParser mqttParser = new IotMqttMessageParser();
registerParser(IotProtocolTypeEnum.MQTT.getCode(), mqttParser);
}
/**
@ -59,7 +60,7 @@ public class DefaultIotProtocolConverter implements IotProtocolConverter {
}
@Override
public IotAlinkMessage convertToStandardMessage(String topic, byte[] payload, String protocol) {
public IotMqttMessage convertToStandardMessage(String topic, byte[] payload, String protocol) {
IotMessageParser parser = parsers.get(protocol);
if (parser == null) {
log.warn(IotLogConstants.Converter.UNSUPPORTED_PROTOCOL, protocol);
@ -108,13 +109,13 @@ public class DefaultIotProtocolConverter implements IotProtocolConverter {
* @param payload 消息负载
* @return 解析后的标准消息，如果无法解析返回 null
*/
public IotAlinkMessage autoConvert(String topic, byte[] payload) {
public IotMqttMessage autoConvert(String topic, byte[] payload) {
// 遍历所有解析器，找到能处理该主题的解析器
for (Map.Entry<String, IotMessageParser> entry : parsers.entrySet()) {
IotMessageParser parser = entry.getValue();
if (parser.canHandle(topic)) {
try {
IotAlinkMessage message = parser.parse(topic, payload);
IotMqttMessage message = parser.parse(topic, payload);
if (message != null) {
log.debug(IotLogConstants.Converter.AUTO_SELECT_PROTOCOL, entry.getKey(), topic);
return message;

4
yudao-module-iot/yudao-module-iot-protocol/src/main/java/cn/iocoder/yudao/module/iot/protocol/enums/IotProtocolTypeEnum.java
View File

@ -13,9 +13,9 @@ import lombok.Getter;
public enum IotProtocolTypeEnum {
/**
* Alink 协议（阿里云物联网协议）
* MQTT 协议（默认实现）
*/
ALINK("alink", "Alink 协议"),
MQTT("mqtt", "MQTT 协议"),
/**
* MQTT 原始协议

2
yudao-module-iot/yudao-module-iot-protocol/src/main/java/cn/iocoder/yudao/module/iot/protocol/message/IotMessageParser.java
View File

@ -16,7 +16,7 @@ public interface IotMessageParser {
* @param payload 消息负载
* @return 解析后的标准消息，如果解析失败返回 null
*/
IotAlinkMessage parse(String topic, byte[] payload);
IotMqttMessage parse(String topic, byte[] payload);
/**
* 格式化响应消息

38
yudao-module-iot/yudao-module-iot-protocol/src/main/java/cn/iocoder/yudao/module/iot/protocol/message/IotAlinkMessage.java → yudao-module-iot/yudao-module-iot-protocol/src/main/java/cn/iocoder/yudao/module/iot/protocol/message/IotMqttMessage.java
View File

@ -8,16 +8,16 @@ import lombok.Data;
import java.util.Map;
/**
* IoT Alink 消息模型
* IoT MQTT 消息模型
* <p>
* 基于阿里云 Alink 协议规范实现的标准消息格式
* 基于 MQTT 协议规范实现的标准消息格式，支持设备属性、事件、服务调用等标准功能
*
* @author haohao
* @see <a href="https://help.aliyun.com/zh/iot/user-guide/alink-protocol-1">阿里云物联网 —— Alink 协议</a>
* @see <a href="https://mqtt.org/">MQTT 协议官方规范</a>
*/
@Data
@Builder
public class IotAlinkMessage {
public class IotMqttMessage {
/**
* 消息 ID
@ -69,11 +69,11 @@ public class IotAlinkMessage {
* @param requestId 请求 ID，为空时自动生成
* @param serviceIdentifier 服务标识符
* @param params 服务参数
* @return Alink 消息对象
* @return MQTT 消息对象
*/
public static IotAlinkMessage createServiceInvokeMessage(String requestId, String serviceIdentifier,
public static IotMqttMessage createServiceInvokeMessage(String requestId, String serviceIdentifier,
Map<String, Object> params) {
return IotAlinkMessage.builder()
return IotMqttMessage.builder()
.id(requestId != null ? requestId : generateRequestId())
.method("thing.service." + serviceIdentifier)
.params(params)
@ -85,10 +85,10 @@ public class IotAlinkMessage {
*
* @param requestId 请求 ID，为空时自动生成
* @param properties 设备属性
* @return Alink 消息对象
* @return MQTT 消息对象
*/
public static IotAlinkMessage createPropertySetMessage(String requestId, Map<String, Object> properties) {
return IotAlinkMessage.builder()
public static IotMqttMessage createPropertySetMessage(String requestId, Map<String, Object> properties) {
return IotMqttMessage.builder()
.id(requestId != null ? requestId : generateRequestId())
.method("thing.service.property.set")
.params(properties)
@ -100,13 +100,13 @@ public class IotAlinkMessage {
*
* @param requestId 请求 ID，为空时自动生成
* @param identifiers 要获取的属性标识符列表
* @return Alink 消息对象
* @return MQTT 消息对象
*/
public static IotAlinkMessage createPropertyGetMessage(String requestId, String[] identifiers) {
public static IotMqttMessage createPropertyGetMessage(String requestId, String[] identifiers) {
JSONObject params = new JSONObject();
params.set("identifiers", identifiers);
return IotAlinkMessage.builder()
return IotMqttMessage.builder()
.id(requestId != null ? requestId : generateRequestId())
.method("thing.service.property.get")
.params(params)
@ -118,10 +118,10 @@ public class IotAlinkMessage {
*
* @param requestId 请求 ID，为空时自动生成
* @param configs 设备配置
* @return Alink 消息对象
* @return MQTT 消息对象
*/
public static IotAlinkMessage createConfigSetMessage(String requestId, Map<String, Object> configs) {
return IotAlinkMessage.builder()
public static IotMqttMessage createConfigSetMessage(String requestId, Map<String, Object> configs) {
return IotMqttMessage.builder()
.id(requestId != null ? requestId : generateRequestId())
.method("thing.service.config.set")
.params(configs)
@ -133,10 +133,10 @@ public class IotAlinkMessage {
*
* @param requestId 请求 ID，为空时自动生成
* @param otaInfo OTA 升级信息
* @return Alink 消息对象
* @return MQTT 消息对象
*/
public static IotAlinkMessage createOtaUpgradeMessage(String requestId, Map<String, Object> otaInfo) {
return IotAlinkMessage.builder()
public static IotMqttMessage createOtaUpgradeMessage(String requestId, Map<String, Object> otaInfo) {
return IotMqttMessage.builder()
.id(requestId != null ? requestId : generateRequestId())
.method("thing.service.ota.upgrade")
.params(otaInfo)

20
yudao-module-iot/yudao-module-iot-protocol/src/main/java/cn/iocoder/yudao/module/iot/protocol/message/impl/IotHttpMessageParser.java
View File

@ -6,8 +6,8 @@ import cn.hutool.json.JSONUtil;
import cn.iocoder.yudao.module.iot.protocol.constants.IotHttpConstants;
import cn.iocoder.yudao.module.iot.protocol.constants.IotLogConstants;
import cn.iocoder.yudao.module.iot.protocol.constants.IotTopicConstants;
import cn.iocoder.yudao.module.iot.protocol.message.IotAlinkMessage;
import cn.iocoder.yudao.module.iot.protocol.message.IotMessageParser;
import cn.iocoder.yudao.module.iot.protocol.message.IotMqttMessage;
import cn.iocoder.yudao.module.iot.protocol.message.IotStandardResponse;
import lombok.extern.slf4j.Slf4j;
@ -61,7 +61,7 @@ public class IotHttpMessageParser implements IotMessageParser {
public static final String TOPIC_PATH_PREFIX = IotHttpConstants.Path.TOPIC_PREFIX;
@Override
public IotAlinkMessage parse(String topic, byte[] payload) {
public IotMqttMessage parse(String topic, byte[] payload) {
if (payload == null || payload.length == 0) {
log.warn(IotLogConstants.Http.RECEIVED_EMPTY_MESSAGE, topic);
return null;
@ -92,7 +92,7 @@ public class IotHttpMessageParser implements IotMessageParser {
* @param message 认证消息JSON
* @return 标准消息格式
*/
private IotAlinkMessage parseAuthMessage(String message) {
private IotMqttMessage parseAuthMessage(String message) {
if (!JSONUtil.isTypeJSON(message)) {
log.warn(IotLogConstants.Http.AUTH_MESSAGE_NOT_JSON, message);
return null;
@ -121,7 +121,7 @@ public class IotHttpMessageParser implements IotMessageParser {
params.put(IotHttpConstants.AuthField.SIGN_METHOD,
json.getStr(IotHttpConstants.AuthField.SIGN_METHOD, IotHttpConstants.DefaultValue.SIGN_METHOD));
return IotAlinkMessage.builder()
return IotMqttMessage.builder()
.id(generateMessageId())
.method(IotHttpConstants.Method.DEVICE_AUTH)
.version(json.getStr(IotHttpConstants.AuthField.VERSION, IotHttpConstants.DefaultValue.VERSION))
@ -136,7 +136,7 @@ public class IotHttpMessageParser implements IotMessageParser {
* @param message 消息内容
* @return 标准消息格式
*/
private IotAlinkMessage parseDataMessage(String topic, String message) {
private IotMqttMessage parseDataMessage(String topic, String message) {
// 提取实际的主题，去掉 /topic 前缀
String actualTopic = topic.substring(TOPIC_PATH_PREFIX.length()); // 直接移除/topic前缀
@ -156,7 +156,7 @@ public class IotHttpMessageParser implements IotMessageParser {
* @param message JSON消息
* @return 标准消息格式
*/
private IotAlinkMessage parseJsonDataMessage(String topic, String message) {
private IotMqttMessage parseJsonDataMessage(String topic, String message) {
JSONObject json = JSONUtil.parseObj(message);
// 生成消息ID
@ -181,7 +181,7 @@ public class IotHttpMessageParser implements IotMessageParser {
paramsMap.put(IotHttpConstants.MessageField.DATA, params);
}
return IotAlinkMessage.builder()
return IotMqttMessage.builder()
.id(messageId)
.method(method)
.version(json.getStr(IotHttpConstants.MessageField.VERSION,
@ -197,11 +197,11 @@ public class IotHttpMessageParser implements IotMessageParser {
* @param message 原始消息
* @return 标准消息格式
*/
private IotAlinkMessage parseRawDataMessage(String topic, String message) {
private IotMqttMessage parseRawDataMessage(String topic, String message) {
Map<String, Object> params = new HashMap<>();
params.put(IotHttpConstants.MessageField.DATA, message);
return IotAlinkMessage.builder()
return IotMqttMessage.builder()
.id(generateMessageId())
.method(inferMethodFromTopic(topic))
.version(IotHttpConstants.DefaultValue.MESSAGE_VERSION)
@ -263,7 +263,7 @@ public class IotHttpMessageParser implements IotMessageParser {
* @return 消息ID
*/
private String generateMessageId() {
return IotAlinkMessage.generateRequestId();
return IotMqttMessage.generateRequestId();
}
@Override

31
yudao-module-iot/yudao-module-iot-protocol/src/main/java/cn/iocoder/yudao/module/iot/protocol/message/impl/IotAlinkMessageParser.java → yudao-module-iot/yudao-module-iot-protocol/src/main/java/cn/iocoder/yudao/module/iot/protocol/message/impl/IotMqttMessageParser.java
View File

@ -4,8 +4,8 @@ import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
import cn.iocoder.yudao.module.iot.protocol.message.IotAlinkMessage;
import cn.iocoder.yudao.module.iot.protocol.message.IotMessageParser;
import cn.iocoder.yudao.module.iot.protocol.message.IotMqttMessage;
import cn.iocoder.yudao.module.iot.protocol.message.IotStandardResponse;
import cn.iocoder.yudao.module.iot.protocol.util.IotTopicUtils;
import lombok.extern.slf4j.Slf4j;
@ -14,26 +14,26 @@ import java.nio.charset.StandardCharsets;
import java.util.Map;
/**
* IoT Alink 协议消息解析器实现
* IoT MQTT 协议消息解析器实现
* <p>
* 基于阿里云 Alink 协议规范实现的消息解析器
* 基于 MQTT 协议规范实现的消息解析器，支持设备属性、事件、服务调用等标准功能
*
* @author haohao
*/
@Slf4j
public class IotAlinkMessageParser implements IotMessageParser {
public class IotMqttMessageParser implements IotMessageParser {
@Override
public IotAlinkMessage parse(String topic, byte[] payload) {
public IotMqttMessage parse(String topic, byte[] payload) {
if (payload == null || payload.length == 0) {
log.warn("[Alink] 收到空消息内容, topic={}", topic);
log.warn("[MQTT] 收到空消息内容, topic={}", topic);
return null;
}
try {
String message = new String(payload, StandardCharsets.UTF_8);
if (!JSONUtil.isTypeJSON(message)) {
log.warn("[Alink] 收到非JSON格式消息, topic={}, message={}", topic, message);
log.warn("[MQTT] 收到非JSON格式消息, topic={}, message={}", topic, message);
return null;
}
@ -45,20 +45,21 @@ public class IotAlinkMessageParser implements IotMessageParser {
// 尝试从 topic 中解析方法
method = IotTopicUtils.parseMethodFromTopic(topic);
if (StrUtil.isBlank(method)) {
log.warn("[Alink] 无法确定消息方法, topic={}, message={}", topic, message);
log.warn("[MQTT] 无法确定消息方法, topic={}, message={}", topic, message);
return null;
}
}
@SuppressWarnings("unchecked")
Map<String, Object> params = (Map<String, Object>) json.getObj("params", Map.class);
return IotAlinkMessage.builder()
return IotMqttMessage.builder()
.id(id)
.method(method)
.version(json.getStr("version", "1.0"))
.params(params)
.build();
} catch (Exception e) {
log.error("[Alink] 解析消息失败, topic={}", topic, e);
log.error("[MQTT] 解析消息失败, topic={}", topic, e);
return null;
}
}
@ -69,14 +70,18 @@ public class IotAlinkMessageParser implements IotMessageParser {
String json = JsonUtils.toJsonString(response);
return json.getBytes(StandardCharsets.UTF_8);
} catch (Exception e) {
log.error("[Alink] 格式化响应失败", e);
log.error("[MQTT] 格式化响应失败", e);
return new byte[0];
}
}
@Override
public boolean canHandle(String topic) {
// Alink 协议处理所有系统主题
return topic != null && topic.startsWith("/sys/");
// MQTT 协议支持更多主题格式
return topic != null && (
topic.startsWith("/sys/") || // 兼容现有系统主题
topic.startsWith("/mqtt/") || // 新的通用 MQTT 主题
topic.startsWith("/device/") // 设备主题
);
}
}

18
yudao-module-iot/yudao-module-iot-protocol/src/test/java/cn/iocoder/yudao/module/iot/protocol/config/IotProtocolAutoConfigurationTest.java
View File

@ -3,8 +3,8 @@ package cn.iocoder.yudao.module.iot.protocol.config;
import cn.iocoder.yudao.module.iot.protocol.convert.IotProtocolConverter;
import cn.iocoder.yudao.module.iot.protocol.enums.IotProtocolTypeEnum;
import cn.iocoder.yudao.module.iot.protocol.message.IotMessageParser;
import cn.iocoder.yudao.module.iot.protocol.message.impl.IotAlinkMessageParser;
import cn.iocoder.yudao.module.iot.protocol.message.impl.IotHttpMessageParser;
import cn.iocoder.yudao.module.iot.protocol.message.impl.IotMqttMessageParser;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@ -25,12 +25,12 @@ class IotProtocolAutoConfigurationTest {
}
@Test
void testIotAlinkMessageParser() {
// 测试 Alink 协议解析器 Bean 创建
IotMessageParser parser = configuration.iotAlinkMessageParser();
void testIotMqttMessageParser() {
// 测试 MQTT 协议解析器 Bean 创建
IotMessageParser parser = configuration.iotMqttMessageParser();
assertNotNull(parser);
assertInstanceOf(IotAlinkMessageParser.class, parser);
assertInstanceOf(IotMqttMessageParser.class, parser);
}
@Test
@ -45,16 +45,16 @@ class IotProtocolAutoConfigurationTest {
@Test
void testIotProtocolConverter() {
// 创建解析器实例
IotMessageParser alinkParser = configuration.iotAlinkMessageParser();
IotMessageParser mqttParser = configuration.iotMqttMessageParser();
IotMessageParser httpParser = configuration.iotHttpMessageParser();
// 测试协议转换器 Bean 创建
IotProtocolConverter converter = configuration.iotProtocolConverter(alinkParser, httpParser);
IotProtocolConverter converter = configuration.iotProtocolConverter(mqttParser, httpParser);
assertNotNull(converter);
// 验证支持的协议
assertTrue(converter.supportsProtocol(IotProtocolTypeEnum.ALINK.getCode()));
assertTrue(converter.supportsProtocol(IotProtocolTypeEnum.MQTT.getCode()));
assertTrue(converter.supportsProtocol(IotProtocolTypeEnum.HTTP.getCode()));
// 验证支持的协议数量
@ -65,7 +65,7 @@ class IotProtocolAutoConfigurationTest {
@Test
void testBeanNameConstants() {
// 测试 Bean 名称常量定义
assertEquals("iotAlinkMessageParser", IotProtocolAutoConfiguration.IOT_ALINK_MESSAGE_PARSER_BEAN_NAME);
assertEquals("iotMqttMessageParser", IotProtocolAutoConfiguration.IOT_MQTT_MESSAGE_PARSER_BEAN_NAME);
assertEquals("iotHttpMessageParser", IotProtocolAutoConfiguration.IOT_HTTP_MESSAGE_PARSER_BEAN_NAME);
}
}

12
yudao-module-iot/yudao-module-iot-protocol/src/test/java/cn/iocoder/yudao/module/iot/protocol/message/impl/IotHttpMessageParserTest.java
View File

@ -1,7 +1,7 @@
package cn.iocoder.yudao.module.iot.protocol.message.impl;
import cn.hutool.json.JSONObject;
import cn.iocoder.yudao.module.iot.protocol.message.IotAlinkMessage;
import cn.iocoder.yudao.module.iot.protocol.message.IotMqttMessage;
import cn.iocoder.yudao.module.iot.protocol.message.IotStandardResponse;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@ -58,7 +58,7 @@ class IotHttpMessageParserTest {
byte[] payload = authMessage.toString().getBytes(StandardCharsets.UTF_8);
// 解析消息
IotAlinkMessage result = parser.parse(topic, payload);
IotMqttMessage result = parser.parse(topic, payload);
// 验证结果
assertNotNull(result);
@ -88,7 +88,7 @@ class IotHttpMessageParserTest {
byte[] payload = authMessage.toString().getBytes(StandardCharsets.UTF_8);
// 解析消息
IotAlinkMessage result = parser.parse(topic, payload);
IotMqttMessage result = parser.parse(topic, payload);
// 验证结果
assertNull(result);
@ -113,7 +113,7 @@ class IotHttpMessageParserTest {
byte[] payload = dataMessage.toString().getBytes(StandardCharsets.UTF_8);
// 解析消息
IotAlinkMessage result = parser.parse(topic, payload);
IotMqttMessage result = parser.parse(topic, payload);
// 验证结果
assertNotNull(result);
@ -132,7 +132,7 @@ class IotHttpMessageParserTest {
byte[] payload = rawData.getBytes(StandardCharsets.UTF_8);
// 解析消息
IotAlinkMessage result = parser.parse(topic, payload);
IotMqttMessage result = parser.parse(topic, payload);
// 验证结果
assertNotNull(result);
@ -161,7 +161,7 @@ class IotHttpMessageParserTest {
String rawData = "test data";
byte[] payload = rawData.getBytes(StandardCharsets.UTF_8);
IotAlinkMessage result = parser.parse(topic, payload);
IotMqttMessage result = parser.parse(topic, payload);
assertNotNull(result);
assertEquals(expectedMethod, result.getMethod());
}

190
yudao-module-iot/yudao-module-iot-protocol/src/test/java/cn/iocoder/yudao/module/iot/protocol/message/impl/IotMqttMessageParserTest.java Normal file
View File

@ -0,0 +1,190 @@
package cn.iocoder.yudao.module.iot.protocol.message.impl;
import cn.hutool.json.JSONObject;
import cn.iocoder.yudao.module.iot.protocol.message.IotMqttMessage;
import cn.iocoder.yudao.module.iot.protocol.message.IotStandardResponse;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import static org.junit.jupiter.api.Assertions.*;
/**
* IoT MQTT 消息解析器测试类
*
* @author haohao
*/
class IotMqttMessageParserTest {
private IotMqttMessageParser parser;
@BeforeEach
void setUp() {
parser = new IotMqttMessageParser();
}
@Test
void testParseValidJsonMessage() {
// 构建有效的 JSON 消息
JSONObject message = new JSONObject();
message.set("id", "123456");
message.set("version", "1.0");
message.set("method", "thing.service.property.set");
Map<String, Object> params = new HashMap<>();
params.put("temperature", 25.5);
params.put("humidity", 60.0);
message.set("params", params);
String topic = "/sys/productKey/deviceName/thing/service/property/set";
byte[] payload = message.toString().getBytes(StandardCharsets.UTF_8);
// 解析消息
IotMqttMessage result = parser.parse(topic, payload);
// 验证结果
assertNotNull(result);
assertEquals("123456", result.getId());
assertEquals("1.0", result.getVersion());
assertEquals("thing.service.property.set", result.getMethod());
assertNotNull(result.getParams());
assertEquals(25.5, ((Number) result.getParams().get("temperature")).doubleValue());
assertEquals(60.0, ((Number) result.getParams().get("humidity")).doubleValue());
}
@Test
void testParseMessageWithoutMethod() {
// 构建没有 method 字段的消息，应该从 topic 中解析
JSONObject message = new JSONObject();
message.set("id", "789012");
message.set("version", "1.0");
Map<String, Object> params = new HashMap<>();
params.put("voltage", 3.3);
message.set("params", params);
String topic = "/sys/productKey/deviceName/thing/service/property/set";
byte[] payload = message.toString().getBytes(StandardCharsets.UTF_8);
// 解析消息
IotMqttMessage result = parser.parse(topic, payload);
// 验证结果
assertNotNull(result);
assertEquals("789012", result.getId());
assertEquals("1.0", result.getVersion());
assertNotNull(result.getMethod()); // 应该从 topic 中解析出方法
assertNotNull(result.getParams());
assertEquals(3.3, ((Number) result.getParams().get("voltage")).doubleValue());
}
@Test
void testParseInvalidJsonMessage() {
String topic = "/sys/productKey/deviceName/thing/service/property/set";
byte[] payload = "invalid json".getBytes(StandardCharsets.UTF_8);
// 解析消息
IotMqttMessage result = parser.parse(topic, payload);
// 验证结果
assertNull(result);
}
@Test
void testParseEmptyPayload() {
String topic = "/sys/productKey/deviceName/thing/service/property/set";
// 测试 null payload
IotMqttMessage result1 = parser.parse(topic, null);
assertNull(result1);
// 测试空 payload
IotMqttMessage result2 = parser.parse(topic, new byte[0]);
assertNull(result2);
}
@Test
void testFormatResponse() {
// 创建标准响应
IotStandardResponse response = IotStandardResponse.success("123456", "property.set", null);
// 格式化响应
byte[] result = parser.formatResponse(response);
// 验证结果
assertNotNull(result);
assertTrue(result.length > 0);
// 验证 JSON 格式
String jsonString = new String(result, StandardCharsets.UTF_8);
assertTrue(jsonString.contains("123456"));
assertTrue(jsonString.contains("property.set"));
}
@Test
void testCanHandle() {
// 测试支持的主题格式
assertTrue(parser.canHandle("/sys/productKey/deviceName/thing/service/property/set"));
assertTrue(parser.canHandle("/mqtt/productKey/deviceName/property/set"));
assertTrue(parser.canHandle("/device/productKey/deviceName/data"));
// 测试不支持的主题格式
assertFalse(parser.canHandle("/http/device/productKey/deviceName/property/set"));
assertFalse(parser.canHandle("/unknown/topic"));
assertFalse(parser.canHandle(null));
assertFalse(parser.canHandle(""));
}
@Test
void testParseMqttTopicFormat() {
// 测试新的 MQTT 主题格式
JSONObject message = new JSONObject();
message.set("id", "mqtt001");
message.set("version", "1.0");
message.set("method", "device.property.report");
Map<String, Object> params = new HashMap<>();
params.put("signal", 85);
message.set("params", params);
String topic = "/mqtt/productKey/deviceName/property/report";
byte[] payload = message.toString().getBytes(StandardCharsets.UTF_8);
// 解析消息
IotMqttMessage result = parser.parse(topic, payload);
// 验证结果
assertNotNull(result);
assertEquals("mqtt001", result.getId());
assertEquals("device.property.report", result.getMethod());
assertEquals(85, ((Number) result.getParams().get("signal")).intValue());
}
@Test
void testParseDeviceTopicFormat() {
// 测试设备主题格式
JSONObject message = new JSONObject();
message.set("id", "device001");
message.set("version", "1.0");
message.set("method", "sensor.data");
Map<String, Object> params = new HashMap<>();
params.put("timestamp", System.currentTimeMillis());
message.set("params", params);
String topic = "/device/productKey/deviceName/sensor/data";
byte[] payload = message.toString().getBytes(StandardCharsets.UTF_8);
// 解析消息
IotMqttMessage result = parser.parse(topic, payload);
// 验证结果
assertNotNull(result);
assertEquals("device001", result.getId());
assertEquals("sensor.data", result.getMethod());
assertNotNull(result.getParams().get("timestamp"));
}
}