diff --git a/yudao-module-iot/yudao-module-iot-biz/pom.xml b/yudao-module-iot/yudao-module-iot-biz/pom.xml
index a0fe16d..02644cb 100644
--- a/yudao-module-iot/yudao-module-iot-biz/pom.xml
+++ b/yudao-module-iot/yudao-module-iot-biz/pom.xml
@@ -101,18 +101,6 @@
spring-boot-starter-amqp
true
-
-
-
-
-
-
-
-
-
-
-
-
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/api/device/IoTDeviceApiImpl.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/api/device/IoTDeviceApiImpl.java
index db0a862..ee4680e 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/api/device/IoTDeviceApiImpl.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/api/device/IoTDeviceApiImpl.java
@@ -58,7 +58,7 @@ public class IoTDeviceApiImpl implements IotDeviceCommonApi {
return success(BeanUtils.toBean(device, IotDeviceRespDTO.class, deviceDTO -> {
IotProductDO product = productService.getProductFromCache(deviceDTO.getProductId());
if (product != null) {
- deviceDTO.setCodecType(product.getCodecType());
+ deviceDTO.setProtocolType(product.getProtocolType()).setSerializeType(product.getSerializeType());
}
}));
}
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/product/vo/product/IotProductRespVO.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/product/vo/product/IotProductRespVO.java
index ffc92a2..302b072 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/product/vo/product/IotProductRespVO.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/product/vo/product/IotProductRespVO.java
@@ -67,10 +67,15 @@ public class IotProductRespVO {
@DictFormat(DictTypeConstants.NET_TYPE)
private Integer netType;
- @Schema(description = "数据格式", requiredMode = Schema.RequiredMode.REQUIRED, example = "2")
- @ExcelProperty(value = "数据格式", converter = DictConvert.class)
- @DictFormat(DictTypeConstants.CODEC_TYPE)
- private String codecType;
+ @Schema(description = "协议类型", requiredMode = Schema.RequiredMode.REQUIRED, example = "mqtt")
+ @ExcelProperty(value = "协议类型", converter = DictConvert.class)
+ @DictFormat(DictTypeConstants.PROTOCOL_TYPE)
+ private String protocolType;
+
+ @Schema(description = "序列化类型", requiredMode = Schema.RequiredMode.REQUIRED, example = "json")
+ @ExcelProperty(value = "序列化类型", converter = DictConvert.class)
+ @DictFormat(DictTypeConstants.SERIALIZE_TYPE)
+ private String serializeType;
@Schema(description = "创建时间", requiredMode = Schema.RequiredMode.REQUIRED)
@ExcelProperty("创建时间")
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/product/vo/product/IotProductSaveReqVO.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/product/vo/product/IotProductSaveReqVO.java
index 08c636f..fceede0 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/product/vo/product/IotProductSaveReqVO.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/product/vo/product/IotProductSaveReqVO.java
@@ -1,6 +1,8 @@
package cn.iocoder.yudao.module.iot.controller.admin.product.vo.product;
import cn.iocoder.yudao.framework.common.validation.InEnum;
+import cn.iocoder.yudao.module.iot.core.enums.IotProtocolTypeEnum;
+import cn.iocoder.yudao.module.iot.core.enums.IotSerializeTypeEnum;
import cn.iocoder.yudao.module.iot.enums.product.IotNetTypeEnum;
import cn.iocoder.yudao.module.iot.enums.product.IotProductDeviceTypeEnum;
import io.swagger.v3.oas.annotations.media.Schema;
@@ -44,9 +46,15 @@ public class IotProductSaveReqVO {
@InEnum(value = IotNetTypeEnum.class, message = "联网方式必须是 {value}")
private Integer netType;
- @Schema(description = "数据格式", requiredMode = Schema.RequiredMode.REQUIRED, example = "2")
- @NotEmpty(message = "数据格式不能为空")
- private String codecType;
+ @Schema(description = "协议类型", requiredMode = Schema.RequiredMode.REQUIRED, example = "mqtt")
+ @InEnum(value = IotProtocolTypeEnum.class, message = "协议类型必须是 {value}")
+ @NotEmpty(message = "协议类型不能为空")
+ private String protocolType;
+
+ @Schema(description = "序列化类型", requiredMode = Schema.RequiredMode.REQUIRED, example = "json")
+ @InEnum(value = IotSerializeTypeEnum.class, message = "序列化类型必须是 {value}")
+ @NotEmpty(message = "序列化类型不能为空")
+ private String serializeType;
@Schema(description = "是否开启动态注册", example = "false")
@NotNull(message = "是否开启动态注册不能为空")
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/product/IotProductDO.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/product/IotProductDO.java
index e296b35..a1a77fc 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/product/IotProductDO.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/product/IotProductDO.java
@@ -78,12 +78,16 @@ public class IotProductDO extends TenantBaseDO {
*/
private Integer netType;
/**
- * 数据格式(编解码器类型)
+ * 协议类型
*
- * 字典 {@link cn.iocoder.yudao.module.iot.enums.DictTypeConstants#CODEC_TYPE}
- *
- * 目的:用于 gateway-server 解析消息格式
+ * 枚举 {@link cn.iocoder.yudao.module.iot.core.enums.IotProtocolTypeEnum}
*/
- private String codecType;
+ private String protocolType;
+ /**
+ * 序列化类型
+ *
+ * 枚举 {@link cn.iocoder.yudao.module.iot.core.enums.IotSerializeTypeEnum}
+ */
+ private String serializeType;
}
\ 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/DictTypeConstants.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/DictTypeConstants.java
index 4f07ddf..cf6bec1 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/DictTypeConstants.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/DictTypeConstants.java
@@ -8,8 +8,8 @@ package cn.iocoder.yudao.module.iot.enums;
public class DictTypeConstants {
public static final String NET_TYPE = "iot_net_type";
- public static final String LOCATION_TYPE = "iot_location_type";
- public static final String CODEC_TYPE = "iot_codec_type";
+ public static final String PROTOCOL_TYPE = "iot_protocol_type";
+ public static final String SERIALIZE_TYPE = "iot_serialize_type";
public static final String PRODUCT_STATUS = "iot_product_status";
public static final String PRODUCT_DEVICE_TYPE = "iot_product_device_type";
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/IotDeviceServiceImpl.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/IotDeviceServiceImpl.java
index 4ec70e0..148dd07 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/IotDeviceServiceImpl.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/IotDeviceServiceImpl.java
@@ -29,6 +29,7 @@ import cn.iocoder.yudao.module.iot.core.topic.topo.IotDeviceTopoChangeReqDTO;
import cn.iocoder.yudao.module.iot.core.topic.topo.IotDeviceTopoDeleteReqDTO;
import cn.iocoder.yudao.module.iot.core.topic.topo.IotDeviceTopoGetRespDTO;
import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils;
+import cn.iocoder.yudao.module.iot.core.util.IotProductAuthUtils;
import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceGroupDO;
import cn.iocoder.yudao.module.iot.dal.dataobject.product.IotProductDO;
@@ -819,8 +820,9 @@ public class IotDeviceServiceImpl implements IotDeviceService {
if (BooleanUtil.isFalse(product.getRegisterEnabled())) {
throw exception(DEVICE_REGISTER_DISABLED);
}
- // 1.3 验证 productSecret
- if (ObjUtil.notEqual(product.getProductSecret(), reqDTO.getProductSecret())) {
+ // 1.3 【重要!!!】验证签名
+ if (!IotProductAuthUtils.verifySign(reqDTO.getProductKey(), reqDTO.getDeviceName(),
+ product.getProductSecret(), reqDTO.getSign())) {
throw exception(DEVICE_REGISTER_SECRET_INVALID);
}
return TenantUtils.execute(product.getTenantId(), () -> {
diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/biz/dto/IotDeviceRespDTO.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/biz/dto/IotDeviceRespDTO.java
index add1167..8ad2c5b 100644
--- a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/biz/dto/IotDeviceRespDTO.java
+++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/biz/dto/IotDeviceRespDTO.java
@@ -34,8 +34,12 @@ public class IotDeviceRespDTO {
*/
private Long productId;
/**
- * 编解码器类型
+ * 协议类型
*/
- private String codecType;
+ private String protocolType;
+ /**
+ * 序列化类型
+ */
+ private String serializeType;
}
\ No newline at end of file
diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/IotProtocolTypeEnum.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/IotProtocolTypeEnum.java
new file mode 100644
index 0000000..5fbd713
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/IotProtocolTypeEnum.java
@@ -0,0 +1,46 @@
+package cn.iocoder.yudao.module.iot.core.enums;
+
+import cn.hutool.core.util.ArrayUtil;
+import cn.iocoder.yudao.framework.common.core.ArrayValuable;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+import java.util.Arrays;
+
+/**
+ * IoT 协议类型枚举
+ *
+ * 用于定义传输层协议类型
+ *
+ * @author 芋道源码
+ */
+@RequiredArgsConstructor
+@Getter
+public enum IotProtocolTypeEnum implements ArrayValuable {
+
+ TCP("tcp"),
+ UDP("udp"),
+ WEBSOCKET("websocket"),
+ HTTP("http"),
+ MQTT("mqtt"),
+ EMQX("emqx"),
+ COAP("coap"),
+ MODBUS_TCP("modbus_tcp");
+
+ public static final String[] ARRAYS = Arrays.stream(values()).map(IotProtocolTypeEnum::getType).toArray(String[]::new);
+
+ /**
+ * 类型
+ */
+ private final String type;
+
+ @Override
+ public String[] array() {
+ return ARRAYS;
+ }
+
+ public static IotProtocolTypeEnum of(String type) {
+ return ArrayUtil.firstMatch(e -> e.getType().equals(type), values());
+ }
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/IotSerializeTypeEnum.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/IotSerializeTypeEnum.java
new file mode 100644
index 0000000..0f9400f
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/IotSerializeTypeEnum.java
@@ -0,0 +1,40 @@
+package cn.iocoder.yudao.module.iot.core.enums;
+
+import cn.hutool.core.util.ArrayUtil;
+import cn.iocoder.yudao.framework.common.core.ArrayValuable;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+import java.util.Arrays;
+
+/**
+ * IoT 序列化类型枚举
+ *
+ * 用于定义设备消息的序列化格式
+ *
+ * @author 芋道源码
+ */
+@RequiredArgsConstructor
+@Getter
+public enum IotSerializeTypeEnum implements ArrayValuable {
+
+ JSON("json"),
+ BINARY("binary");
+
+ public static final String[] ARRAYS = Arrays.stream(values()).map(IotSerializeTypeEnum::getType).toArray(String[]::new);
+
+ /**
+ * 类型
+ */
+ private final String type;
+
+ @Override
+ public String[] array() {
+ return ARRAYS;
+ }
+
+ public static IotSerializeTypeEnum of(String type) {
+ return ArrayUtil.firstMatch(e -> e.getType().equals(type), values());
+ }
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/IotMessageBus.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/IotMessageBus.java
index c621467..646eb36 100644
--- a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/IotMessageBus.java
+++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/IotMessageBus.java
@@ -24,4 +24,14 @@ public interface IotMessageBus {
*/
void register(IotMessageSubscriber> subscriber);
+ /**
+ * 取消注册消息订阅者
+ *
+ * @param subscriber 订阅者
+ */
+ default void unregister(IotMessageSubscriber> subscriber) {
+ // TODO 芋艿:暂时不实现,需求量不大,但是
+ // throw new UnsupportedOperationException("取消注册消息订阅者功能,尚未实现");
+ }
+
}
\ No newline at end of file
diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/IotMessageSubscriber.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/IotMessageSubscriber.java
index 23a0553..fb5c712 100644
--- a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/IotMessageSubscriber.java
+++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/IotMessageSubscriber.java
@@ -26,4 +26,16 @@ public interface IotMessageSubscriber {
*/
void onMessage(T message);
+ /**
+ * 启动订阅
+ */
+ default void start() {
+ }
+
+ /**
+ * 停止订阅
+ */
+ default void stop() {
+ }
+
}
\ No newline at end of file
diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/mq/message/IotDeviceMessage.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/mq/message/IotDeviceMessage.java
index feed3eb..813b360 100644
--- a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/mq/message/IotDeviceMessage.java
+++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/mq/message/IotDeviceMessage.java
@@ -60,7 +60,7 @@ public class IotDeviceMessage {
*/
private String serverId;
- // ========== codec(编解码)字段 ==========
+ // ========== serialize(序列化)相关字段 ==========
/**
* 请求编号
@@ -94,7 +94,7 @@ public class IotDeviceMessage {
*/
private String msg;
- // ========== 基础方法:只传递"codec(编解码)字段" ==========
+ // ========== 基础方法:只传递"serialize(序列化)相关字段" ==========
public static IotDeviceMessage requestOf(String method) {
return requestOf(null, method, null);
diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/auth/IotDeviceRegisterReqDTO.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/auth/IotDeviceRegisterReqDTO.java
index b8db15f..a77cd42 100644
--- a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/auth/IotDeviceRegisterReqDTO.java
+++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/auth/IotDeviceRegisterReqDTO.java
@@ -27,9 +27,11 @@ public class IotDeviceRegisterReqDTO {
private String deviceName;
/**
- * 产品密钥
+ * 注册签名
+ *
+ * @see cn.iocoder.yudao.module.iot.core.util.IotProductAuthUtils#buildSign(String, String, String)
*/
- @NotEmpty(message = "产品密钥不能为空")
- private String productSecret;
+ @NotEmpty(message = "签名不能为空")
+ private String sign;
}
diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/util/IotProductAuthUtils.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/util/IotProductAuthUtils.java
new file mode 100644
index 0000000..12d1229
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/util/IotProductAuthUtils.java
@@ -0,0 +1,55 @@
+package cn.iocoder.yudao.module.iot.core.util;
+
+import cn.hutool.core.util.StrUtil;
+import cn.hutool.crypto.digest.DigestUtil;
+import cn.hutool.crypto.digest.HmacAlgorithm;
+
+/**
+ * IoT 产品【动态注册】认证工具类
+ *
+ * 用于一型一密场景,使用 productSecret 生成签名
+ *
+ * @author 芋道源码
+ */
+public class IotProductAuthUtils {
+
+ /**
+ * 生成设备动态注册签名
+ *
+ * @param productKey 产品标识
+ * @param deviceName 设备名称
+ * @param productSecret 产品密钥
+ * @return 签名
+ */
+ public static String buildSign(String productKey, String deviceName, String productSecret) {
+ String content = buildContent(productKey, deviceName);
+ return DigestUtil.hmac(HmacAlgorithm.HmacSHA256, StrUtil.utf8Bytes(productSecret))
+ .digestHex(content);
+ }
+
+ /**
+ * 验证设备动态注册签名
+ *
+ * @param productKey 产品标识
+ * @param deviceName 设备名称
+ * @param productSecret 产品密钥
+ * @param sign 待验证的签名
+ * @return 是否验证通过
+ */
+ public static boolean verifySign(String productKey, String deviceName, String productSecret, String sign) {
+ String expectedSign = buildSign(productKey, deviceName, productSecret);
+ return expectedSign.equals(sign);
+ }
+
+ /**
+ * 构建签名内容
+ *
+ * @param productKey 产品标识
+ * @param deviceName 设备名称
+ * @return 签名内容
+ */
+ private static String buildContent(String productKey, String deviceName) {
+ return "deviceName" + deviceName + "productKey" + productKey;
+ }
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/codec/alink/IotAlinkDeviceMessageCodec.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/codec/alink/IotAlinkDeviceMessageCodec.java
deleted file mode 100644
index 5a4e47f..0000000
--- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/codec/alink/IotAlinkDeviceMessageCodec.java
+++ /dev/null
@@ -1,89 +0,0 @@
-package cn.iocoder.yudao.module.iot.gateway.codec.alink;
-
-import cn.hutool.core.lang.Assert;
-import cn.iocoder.yudao.framework.common.pojo.CommonResult;
-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.gateway.codec.IotDeviceMessageCodec;
-import lombok.AllArgsConstructor;
-import lombok.Data;
-import lombok.NoArgsConstructor;
-import org.springframework.stereotype.Component;
-
-/**
- * 阿里云 Alink {@link IotDeviceMessage} 的编解码器
- *
- * @author 芋道源码
- */
-@Component
-public class IotAlinkDeviceMessageCodec implements IotDeviceMessageCodec {
-
- public static final String TYPE = "Alink";
-
- @Data
- @NoArgsConstructor
- @AllArgsConstructor
- private static class AlinkMessage {
-
- public static final String VERSION_1 = "1.0";
-
- /**
- * 消息 ID,且每个消息 ID 在当前设备具有唯一性
- */
- private String id;
-
- /**
- * 版本号
- */
- private String version;
-
- /**
- * 请求方法
- */
- private String method;
-
- /**
- * 请求参数
- */
- private Object params;
-
- /**
- * 响应结果
- */
- private Object data;
- /**
- * 响应错误码
- */
- private Integer code;
- /**
- * 响应提示
- *
- * 特殊:这里阿里云是 message,为了保持和项目的 {@link CommonResult#getMsg()} 一致。
- */
- private String msg;
-
- }
-
- @Override
- public String type() {
- return TYPE;
- }
-
- @Override
- public byte[] encode(IotDeviceMessage message) {
- AlinkMessage alinkMessage = new AlinkMessage(message.getRequestId(), AlinkMessage.VERSION_1,
- message.getMethod(), message.getParams(), message.getData(), message.getCode(), message.getMsg());
- return JsonUtils.toJsonByte(alinkMessage);
- }
-
- @Override
- @SuppressWarnings("DataFlowIssue")
- public IotDeviceMessage decode(byte[] bytes) {
- AlinkMessage alinkMessage = JsonUtils.parseObject(bytes, AlinkMessage.class);
- Assert.notNull(alinkMessage, "消息不能为空");
- Assert.equals(alinkMessage.getVersion(), AlinkMessage.VERSION_1, "消息版本号必须是 1.0");
- return IotDeviceMessage.of(alinkMessage.getId(), alinkMessage.getMethod(), alinkMessage.getParams(),
- alinkMessage.getData(), alinkMessage.getCode(), alinkMessage.getMsg());
- }
-
-}
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/codec/package-info.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/codec/package-info.java
deleted file mode 100644
index e1dae77..0000000
--- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/codec/package-info.java
+++ /dev/null
@@ -1,4 +0,0 @@
-/**
- * 提供设备接入的各种数据(请求、响应)的编解码
- */
-package cn.iocoder.yudao.module.iot.gateway.codec;
\ No newline at end of file
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/codec/simple/package-info.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/codec/simple/package-info.java
deleted file mode 100644
index 5bd676a..0000000
--- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/codec/simple/package-info.java
+++ /dev/null
@@ -1,4 +0,0 @@
-/**
- * TODO @芋艿:实现一个 alink 的 xml 版本
- */
-package cn.iocoder.yudao.module.iot.gateway.codec.simple;
\ No newline at end of file
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/codec/tcp/IotTcpJsonDeviceMessageCodec.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/codec/tcp/IotTcpJsonDeviceMessageCodec.java
deleted file mode 100644
index 7d62ce2..0000000
--- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/codec/tcp/IotTcpJsonDeviceMessageCodec.java
+++ /dev/null
@@ -1,110 +0,0 @@
-package cn.iocoder.yudao.module.iot.gateway.codec.tcp;
-
-import cn.hutool.core.lang.Assert;
-import cn.hutool.core.util.StrUtil;
-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.gateway.codec.IotDeviceMessageCodec;
-import lombok.AllArgsConstructor;
-import lombok.Data;
-import lombok.NoArgsConstructor;
-import org.springframework.stereotype.Component;
-
-/**
- * TCP/UDP JSON 格式 {@link IotDeviceMessage} 编解码器
- *
- * 采用纯 JSON 格式传输,格式如下:
- * {
- * "id": "消息 ID",
- * "method": "消息方法",
- * "params": {...}, // 请求参数
- * "data": {...}, // 响应结果
- * "code": 200, // 响应错误码
- * "msg": "success", // 响应提示
- * "timestamp": 时间戳
- * }
- *
- * @author 芋道源码
- */
-@Component
-public class IotTcpJsonDeviceMessageCodec implements IotDeviceMessageCodec {
-
- public static final String TYPE = "TCP_JSON";
-
- @Data
- @NoArgsConstructor
- @AllArgsConstructor
- private static class TcpJsonMessage {
-
- /**
- * 消息 ID,且每个消息 ID 在当前设备具有唯一性
- */
- private String id;
-
- /**
- * 请求方法
- */
- private String method;
-
- /**
- * 请求参数
- */
- private Object params;
-
- /**
- * 响应结果
- */
- private Object data;
-
- /**
- * 响应错误码
- */
- private Integer code;
-
- /**
- * 响应提示
- */
- private String msg;
-
- /**
- * 时间戳
- */
- private Long timestamp;
-
- }
-
- @Override
- public String type() {
- return TYPE;
- }
-
- @Override
- public byte[] encode(IotDeviceMessage message) {
- TcpJsonMessage tcpJsonMessage = new TcpJsonMessage(
- message.getRequestId(),
- message.getMethod(),
- message.getParams(),
- message.getData(),
- message.getCode(),
- message.getMsg(),
- System.currentTimeMillis());
- return JsonUtils.toJsonByte(tcpJsonMessage);
- }
-
- @Override
- @SuppressWarnings("DataFlowIssue")
- public IotDeviceMessage decode(byte[] bytes) {
- String jsonStr = StrUtil.utf8Str(bytes).trim();
- TcpJsonMessage tcpJsonMessage = JsonUtils.parseObject(jsonStr, TcpJsonMessage.class);
- Assert.notNull(tcpJsonMessage, "消息不能为空");
- Assert.notBlank(tcpJsonMessage.getMethod(), "消息方法不能为空");
- return IotDeviceMessage.of(
- tcpJsonMessage.getId(),
- tcpJsonMessage.getMethod(),
- tcpJsonMessage.getParams(),
- tcpJsonMessage.getData(),
- tcpJsonMessage.getCode(),
- tcpJsonMessage.getMsg());
- }
-
-}
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 a4e93a8..3c62c0d 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
@@ -1,254 +1,23 @@
package cn.iocoder.yudao.module.iot.gateway.config;
-import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
-import cn.iocoder.yudao.module.iot.gateway.protocol.coap.IotCoapDownstreamSubscriber;
-import cn.iocoder.yudao.module.iot.gateway.protocol.coap.IotCoapUpstreamProtocol;
-import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.IotEmqxAuthEventProtocol;
-import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.IotEmqxDownstreamSubscriber;
-import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.IotEmqxUpstreamProtocol;
-import cn.iocoder.yudao.module.iot.gateway.protocol.http.IotHttpDownstreamSubscriber;
-import cn.iocoder.yudao.module.iot.gateway.protocol.http.IotHttpUpstreamProtocol;
-import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttDownstreamSubscriber;
-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.tcp.IotTcpDownstreamSubscriber;
-import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.IotTcpUpstreamProtocol;
-import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.manager.IotTcpConnectionManager;
-import cn.iocoder.yudao.module.iot.gateway.protocol.udp.IotUdpDownstreamSubscriber;
-import cn.iocoder.yudao.module.iot.gateway.protocol.udp.IotUdpUpstreamProtocol;
-import cn.iocoder.yudao.module.iot.gateway.protocol.udp.manager.IotUdpSessionManager;
-import cn.iocoder.yudao.module.iot.gateway.protocol.websocket.IotWebSocketDownstreamSubscriber;
-import cn.iocoder.yudao.module.iot.gateway.protocol.websocket.IotWebSocketUpstreamProtocol;
-import cn.iocoder.yudao.module.iot.gateway.protocol.websocket.manager.IotWebSocketConnectionManager;
-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 cn.iocoder.yudao.module.iot.gateway.protocol.IotProtocolManager;
+import cn.iocoder.yudao.module.iot.gateway.serialize.IotMessageSerializerManager;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
@EnableConfigurationProperties(IotGatewayProperties.class)
-@Slf4j
public class IotGatewayConfiguration {
- /**
- * IoT 网关 HTTP 协议配置类
- */
- @Configuration
- @ConditionalOnProperty(prefix = "yudao.iot.gateway.protocol.http", name = "enabled", havingValue = "true")
- @Slf4j
- public static class HttpProtocolConfiguration {
-
- @Bean(name = "httpVertx", destroyMethod = "close")
- public Vertx httpVertx() {
- return Vertx.vertx();
- }
-
- @Bean
- public IotHttpUpstreamProtocol iotHttpUpstreamProtocol(IotGatewayProperties gatewayProperties,
- @Qualifier("httpVertx") Vertx httpVertx) {
- return new IotHttpUpstreamProtocol(gatewayProperties.getProtocol().getHttp(), httpVertx);
- }
-
- @Bean
- public IotHttpDownstreamSubscriber iotHttpDownstreamSubscriber(IotHttpUpstreamProtocol httpUpstreamProtocol,
- IotMessageBus messageBus) {
- return new IotHttpDownstreamSubscriber(httpUpstreamProtocol, messageBus);
- }
- }
-
- /**
- * IoT 网关 EMQX 协议配置类
- */
- @Configuration
- @ConditionalOnProperty(prefix = "yudao.iot.gateway.protocol.emqx", name = "enabled", havingValue = "true")
- @Slf4j
- public static class EmqxProtocolConfiguration {
-
- @Bean(name = "emqxVertx", destroyMethod = "close")
- public Vertx emqxVertx() {
- return Vertx.vertx();
- }
-
- @Bean
- public IotEmqxAuthEventProtocol iotEmqxAuthEventProtocol(IotGatewayProperties gatewayProperties,
- @Qualifier("emqxVertx") Vertx emqxVertx) {
- return new IotEmqxAuthEventProtocol(gatewayProperties.getProtocol().getEmqx(), emqxVertx);
- }
-
- @Bean
- public IotEmqxUpstreamProtocol iotEmqxUpstreamProtocol(IotGatewayProperties gatewayProperties,
- @Qualifier("emqxVertx") Vertx emqxVertx) {
- return new IotEmqxUpstreamProtocol(gatewayProperties.getProtocol().getEmqx(), emqxVertx);
- }
-
- @Bean
- public IotEmqxDownstreamSubscriber iotEmqxDownstreamSubscriber(IotEmqxUpstreamProtocol mqttUpstreamProtocol,
- IotMessageBus messageBus) {
- return new IotEmqxDownstreamSubscriber(mqttUpstreamProtocol, messageBus);
- }
+ @Bean
+ public IotMessageSerializerManager iotMessageSerializerManager() {
+ return new IotMessageSerializerManager();
}
- /**
- * IoT 网关 TCP 协议配置类
- */
- @Configuration
- @ConditionalOnProperty(prefix = "yudao.iot.gateway.protocol.tcp", name = "enabled", havingValue = "true")
- @Slf4j
- public static class TcpProtocolConfiguration {
-
- @Bean(name = "tcpVertx", destroyMethod = "close")
- public Vertx tcpVertx() {
- return Vertx.vertx();
- }
-
- @Bean
- public IotTcpUpstreamProtocol iotTcpUpstreamProtocol(IotGatewayProperties gatewayProperties,
- IotDeviceService deviceService,
- IotDeviceMessageService messageService,
- IotTcpConnectionManager connectionManager,
- @Qualifier("tcpVertx") Vertx tcpVertx) {
- return new IotTcpUpstreamProtocol(gatewayProperties.getProtocol().getTcp(),
- deviceService, messageService, connectionManager, tcpVertx);
- }
-
- @Bean
- public IotTcpDownstreamSubscriber iotTcpDownstreamSubscriber(IotTcpUpstreamProtocol protocolHandler,
- IotDeviceMessageService messageService,
- IotTcpConnectionManager connectionManager,
- IotMessageBus messageBus) {
- return new IotTcpDownstreamSubscriber(protocolHandler, messageService, connectionManager, messageBus);
- }
-
- }
-
- /**
- * IoT 网关 MQTT 协议配置类
- */
- @Configuration
- @ConditionalOnProperty(prefix = "yudao.iot.gateway.protocol.mqtt", name = "enabled", havingValue = "true")
- @Slf4j
- public static class MqttProtocolConfiguration {
-
- @Bean(name = "mqttVertx", destroyMethod = "close")
- public Vertx mqttVertx() {
- return Vertx.vertx();
- }
-
- @Bean
- public IotMqttUpstreamProtocol iotMqttUpstreamProtocol(IotGatewayProperties gatewayProperties,
- IotDeviceMessageService messageService,
- IotMqttConnectionManager connectionManager,
- @Qualifier("mqttVertx") Vertx mqttVertx) {
- return new IotMqttUpstreamProtocol(gatewayProperties.getProtocol().getMqtt(), messageService,
- connectionManager, mqttVertx);
- }
-
- @Bean
- public IotMqttDownstreamHandler iotMqttDownstreamHandler(IotDeviceMessageService messageService,
- IotMqttConnectionManager connectionManager) {
- return new IotMqttDownstreamHandler(messageService, connectionManager);
- }
-
- @Bean
- public IotMqttDownstreamSubscriber iotMqttDownstreamSubscriber(IotMqttUpstreamProtocol mqttUpstreamProtocol,
- IotMqttDownstreamHandler downstreamHandler,
- IotMessageBus messageBus) {
- return new IotMqttDownstreamSubscriber(mqttUpstreamProtocol, downstreamHandler, messageBus);
- }
-
- }
-
- /**
- * IoT 网关 UDP 协议配置类
- */
- @Configuration
- @ConditionalOnProperty(prefix = "yudao.iot.gateway.protocol.udp", name = "enabled", havingValue = "true")
- @Slf4j
- public static class UdpProtocolConfiguration {
-
- @Bean(name = "udpVertx", destroyMethod = "close")
- public Vertx udpVertx() {
- return Vertx.vertx();
- }
-
- @Bean
- public IotUdpUpstreamProtocol iotUdpUpstreamProtocol(IotGatewayProperties gatewayProperties,
- IotDeviceService deviceService,
- IotDeviceMessageService messageService,
- IotUdpSessionManager sessionManager,
- @Qualifier("udpVertx") Vertx udpVertx) {
- return new IotUdpUpstreamProtocol(gatewayProperties.getProtocol().getUdp(),
- deviceService, messageService, sessionManager, udpVertx);
- }
-
- @Bean
- public IotUdpDownstreamSubscriber iotUdpDownstreamSubscriber(IotUdpUpstreamProtocol protocolHandler,
- IotDeviceMessageService messageService,
- IotUdpSessionManager sessionManager,
- IotMessageBus messageBus) {
- return new IotUdpDownstreamSubscriber(protocolHandler, messageService, sessionManager, messageBus);
- }
-
- }
-
- /**
- * IoT 网关 CoAP 协议配置类
- */
- @Configuration
- @ConditionalOnProperty(prefix = "yudao.iot.gateway.protocol.coap", name = "enabled", havingValue = "true")
- @Slf4j
- public static class CoapProtocolConfiguration {
-
- @Bean
- public IotCoapUpstreamProtocol iotCoapUpstreamProtocol(IotGatewayProperties gatewayProperties) {
- return new IotCoapUpstreamProtocol(gatewayProperties.getProtocol().getCoap());
- }
-
- @Bean
- public IotCoapDownstreamSubscriber iotCoapDownstreamSubscriber(IotCoapUpstreamProtocol coapUpstreamProtocol,
- IotMessageBus messageBus) {
- return new IotCoapDownstreamSubscriber(coapUpstreamProtocol, messageBus);
- }
-
- }
-
- /**
- * IoT 网关 WebSocket 协议配置类
- */
- @Configuration
- @ConditionalOnProperty(prefix = "yudao.iot.gateway.protocol.websocket", name = "enabled", havingValue = "true")
- @Slf4j
- public static class WebSocketProtocolConfiguration {
-
- @Bean(name = "websocketVertx", destroyMethod = "close")
- public Vertx websocketVertx() {
- return Vertx.vertx();
- }
-
- @Bean
- public IotWebSocketUpstreamProtocol iotWebSocketUpstreamProtocol(IotGatewayProperties gatewayProperties,
- IotDeviceService deviceService,
- IotDeviceMessageService messageService,
- IotWebSocketConnectionManager connectionManager,
- @Qualifier("websocketVertx") Vertx websocketVertx) {
- return new IotWebSocketUpstreamProtocol(gatewayProperties.getProtocol().getWebsocket(),
- deviceService, messageService, connectionManager, websocketVertx);
- }
-
- @Bean
- public IotWebSocketDownstreamSubscriber iotWebSocketDownstreamSubscriber(IotWebSocketUpstreamProtocol protocolHandler,
- IotDeviceMessageService messageService,
- IotWebSocketConnectionManager connectionManager,
- IotMessageBus messageBus) {
- return new IotWebSocketDownstreamSubscriber(protocolHandler, messageService, connectionManager, messageBus);
- }
-
+ @Bean
+ public IotProtocolManager iotProtocolManager(IotGatewayProperties gatewayProperties) {
+ return new IotProtocolManager(gatewayProperties);
}
}
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 9a86ee6..6019b0f 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
@@ -1,5 +1,14 @@
package cn.iocoder.yudao.module.iot.gateway.config;
+import cn.iocoder.yudao.module.iot.core.enums.IotProtocolTypeEnum;
+import cn.iocoder.yudao.module.iot.gateway.protocol.coap.IotCoapConfig;
+import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.IotEmqxConfig;
+import cn.iocoder.yudao.module.iot.gateway.protocol.http.IotHttpConfig;
+import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttConfig;
+import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.IotTcpConfig;
+import cn.iocoder.yudao.module.iot.gateway.protocol.udp.IotUdpConfig;
+import cn.iocoder.yudao.module.iot.gateway.protocol.websocket.IotWebSocketConfig;
+import jakarta.validation.Valid;
import jakarta.validation.constraints.NotEmpty;
import jakarta.validation.constraints.NotNull;
import lombok.Data;
@@ -24,9 +33,9 @@ public class IotGatewayProperties {
private TokenProperties token;
/**
- * 协议配置
+ * 协议实例列表
*/
- private ProtocolProperties protocol;
+ private List protocols;
@Data
public static class RpcProperties {
@@ -65,581 +74,145 @@ public class IotGatewayProperties {
}
+ /**
+ * 协议实例配置
+ */
@Data
public static class ProtocolProperties {
/**
- * HTTP 组件配置
- */
- private HttpProperties http;
-
- /**
- * EMQX 组件配置
- */
- private EmqxProperties emqx;
-
- /**
- * TCP 组件配置
+ * 协议实例 ID,如 "http-alink"、"tcp-binary"
*/
- private TcpProperties tcp;
-
- /**
- * MQTT 组件配置
- */
- private MqttProperties mqtt;
-
+ @NotEmpty(message = "协议实例 ID 不能为空")
+ private String id;
/**
- * MQTT WebSocket 组件配置
+ * 是否启用
*/
- private MqttWsProperties mqttWs;
-
+ @NotNull(message = "是否启用不能为空")
+ private Boolean enabled = true;
/**
- * UDP 组件配置
+ * 协议类型
+ *
+ * @see cn.iocoder.yudao.module.iot.core.enums.IotProtocolTypeEnum
*/
- private UdpProperties udp;
-
- /**
- * CoAP 组件配置
- */
- private CoapProperties coap;
-
- /**
- * WebSocket 组件配置
- */
- private WebSocketProperties websocket;
-
- }
-
- @Data
- public static class HttpProperties {
-
- /**
- * 是否开启
- */
- @NotNull(message = "是否开启不能为空")
- private Boolean enabled;
+ @NotEmpty(message = "协议类型不能为空")
+ private String protocol;
/**
* 服务端口
- */
- private Integer serverPort;
-
- /**
- * 是否开启 SSL
- */
- @NotNull(message = "是否开启 SSL 不能为空")
- private Boolean sslEnabled = false;
-
- /**
- * SSL 证书路径
- */
- private String sslKeyPath;
- /**
- * SSL 证书路径
- */
- private String sslCertPath;
-
- }
-
- @Data
- public static class EmqxProperties {
-
- /**
- * 是否开启
- */
- @NotNull(message = "是否开启不能为空")
- private Boolean enabled;
-
- /**
- * HTTP 服务端口(默认:8090)
- */
- private Integer httpPort = 8090;
-
- /**
- * MQTT 服务器地址
- */
- @NotEmpty(message = "MQTT 服务器地址不能为空")
- private String mqttHost;
-
- /**
- * MQTT 服务器端口(默认:1883)
- */
- @NotNull(message = "MQTT 服务器端口不能为空")
- private Integer mqttPort = 1883;
-
- /**
- * MQTT 用户名
- */
- @NotEmpty(message = "MQTT 用户名不能为空")
- private String mqttUsername;
-
- /**
- * MQTT 密码
- */
- @NotEmpty(message = "MQTT 密码不能为空")
- private String mqttPassword;
-
- /**
- * MQTT 客户端的 SSL 开关
- */
- @NotNull(message = "MQTT 是否开启 SSL 不能为空")
- private Boolean mqttSsl = false;
-
- /**
- * MQTT 客户端 ID(如果为空,系统将自动生成)
- */
- @NotEmpty(message = "MQTT 客户端 ID 不能为空")
- private String mqttClientId;
-
- /**
- * MQTT 订阅的主题
- */
- @NotEmpty(message = "MQTT 主题不能为空")
- private List<@NotEmpty(message = "MQTT 主题不能为空") String> mqttTopics;
-
- /**
- * 默认 QoS 级别
*
- * 0 - 最多一次
- * 1 - 至少一次
- * 2 - 刚好一次
- */
- private Integer mqttQos = 1;
-
- /**
- * 连接超时时间(秒)
- */
- private Integer connectTimeoutSeconds = 10;
-
- /**
- * 重连延迟时间(毫秒)
- */
- private Long reconnectDelayMs = 5000L;
-
- /**
- * 是否启用 Clean Session (清理会话)
- * true: 每次连接都是新会话,Broker 不保留离线消息和订阅关系。
- * 对于网关这类“永远在线”且会主动重新订阅的应用,建议为 true。
- */
- private Boolean cleanSession = true;
-
- /**
- * 心跳间隔(秒)
- * 用于保持连接活性,及时发现网络中断。
- */
- private Integer keepAliveIntervalSeconds = 60;
-
- /**
- * 最大未确认消息队列大小
- * 限制已发送但未收到 Broker 确认的 QoS 1/2 消息数量,用于流量控制。
- */
- private Integer maxInflightQueue = 10000;
-
- /**
- * 是否信任所有 SSL 证书
- * 警告:此配置会绕过证书验证,仅建议在开发和测试环境中使用!
- * 在生产环境中,应设置为 false,并配置正确的信任库。
- */
- private Boolean trustAll = false;
-
- /**
- * 遗嘱消息配置 (用于网关异常下线时通知其他系统)
- */
- private final Will will = new Will();
-
- /**
- * 高级 SSL/TLS 配置 (用于生产环境)
+ * 不同协议含义不同:
+ * 1. TCP/UDP/HTTP/WebSocket/MQTT/CoAP:对应网关自身监听的服务端口
+ * 2. EMQX:对应网关提供给 EMQX 回调的 HTTP Hook 端口(/mqtt/auth、/mqtt/acl、/mqtt/event)
*/
- private final Ssl sslOptions = new Ssl();
-
+ @NotNull(message = "服务端口不能为空")
+ private Integer port;
/**
- * 遗嘱消息 (Last Will and Testament)
+ * 序列化类型(可选)
+ *
+ * @see cn.iocoder.yudao.module.iot.core.enums.IotSerializeTypeEnum
+ *
+ * 为什么是可选的呢?
+ * 1. {@link IotProtocolTypeEnum#HTTP}、{@link IotProtocolTypeEnum#COAP} 协议,目前强制是 JSON 格式
+ * 2. {@link IotProtocolTypeEnum#EMQX} 协议,目前支持根据产品(设备)配置的序列化类型来解析
*/
- @Data
- public static class Will {
-
- /**
- * 是否启用遗嘱消息
- */
- private boolean enabled = false;
- /**
- * 遗嘱消息主题
- */
- private String topic;
- /**
- * 遗嘱消息内容
- */
- private String payload;
- /**
- * 遗嘱消息 QoS 等级
- */
- private Integer qos = 1;
- /**
- * 遗嘱消息是否作为保留消息发布
- */
- private boolean retain = true;
+ private String serialize;
- }
+ // ========== SSL 配置 ==========
/**
- * 高级 SSL/TLS 配置
+ * SSL 配置(可选,配置文件中不配置则为 null)
*/
- @Data
- public static class Ssl {
+ @Valid
+ private SslConfig ssl;
- /**
- * 密钥库(KeyStore)路径,例如:classpath:certs/client.jks
- * 包含客户端自己的证书和私钥,用于向服务端证明身份(双向认证)。
- */
- private String keyStorePath;
- /**
- * 密钥库密码
- */
- private String keyStorePassword;
- /**
- * 信任库(TrustStore)路径,例如:classpath:certs/trust.jks
- * 包含服务端信任的 CA 证书,用于验证服务端的身份,防止中间人攻击。
- */
- private String trustStorePath;
- /**
- * 信任库密码
- */
- private String trustStorePassword;
-
- }
-
- }
-
- @Data
- public static class TcpProperties {
+ // ========== 各协议配置 ==========
/**
- * 是否开启
+ * HTTP 协议配置
*/
- @NotNull(message = "是否开启不能为空")
- private Boolean enabled;
-
+ @Valid
+ private IotHttpConfig http;
/**
- * 服务器端口
+ * WebSocket 协议配置
*/
- private Integer port = 8091;
+ @Valid
+ private IotWebSocketConfig websocket;
/**
- * 心跳超时时间(毫秒)
+ * TCP 协议配置
*/
- private Long keepAliveTimeoutMs = 30000L;
-
+ @Valid
+ private IotTcpConfig tcp;
/**
- * 最大连接数
+ * UDP 协议配置
*/
- private Integer maxConnections = 1000;
+ @Valid
+ private IotUdpConfig udp;
/**
- * 是否启用SSL
+ * CoAP 协议配置
*/
- private Boolean sslEnabled = false;
+ @Valid
+ private IotCoapConfig coap;
/**
- * SSL证书路径
+ * MQTT 协议配置
*/
- private String sslCertPath;
-
+ @Valid
+ private IotMqttConfig mqtt;
/**
- * SSL私钥路径
+ * EMQX 协议配置
*/
- private String sslKeyPath;
+ @Valid
+ private IotEmqxConfig emqx;
}
+ /**
+ * SSL 配置
+ */
@Data
- public static class MqttProperties {
-
- /**
- * 是否开启
- */
- @NotNull(message = "是否开启不能为空")
- private Boolean enabled;
-
- /**
- * 服务器端口
- */
- private Integer port = 1883;
-
- /**
- * 最大消息大小(字节)
- */
- private Integer maxMessageSize = 8192;
-
- /**
- * 连接超时时间(秒)
- */
- private Integer connectTimeoutSeconds = 60;
- /**
- * 保持连接超时时间(秒)
- */
- private Integer keepAliveTimeoutSeconds = 300;
+ public static class SslConfig {
/**
* 是否启用 SSL
*/
- private Boolean sslEnabled = false;
- /**
- * SSL 配置
- */
- private SslOptions sslOptions = new SslOptions();
-
- /**
- * 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;
-
- }
-
- }
-
- @Data
- public static class MqttWsProperties {
+ @NotNull(message = "是否启用 SSL 不能为空")
+ private Boolean ssl = false;
/**
- * 是否开启
- */
- @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;
-
- }
-
- }
-
- @Data
- public static class UdpProperties {
-
- /**
- * 是否开启
- */
- @NotNull(message = "是否开启不能为空")
- private Boolean enabled;
-
- /**
- * 服务端口(默认 8093)
- */
- private Integer port = 8093;
-
- /**
- * 接收缓冲区大小(默认 64KB)
+ * SSL 证书路径
*/
- private Integer receiveBufferSize = 65536;
+ @NotEmpty(message = "SSL 证书路径不能为空")
+ private String sslCertPath;
/**
- * 发送缓冲区大小(默认 64KB)
+ * SSL 私钥路径
*/
- private Integer sendBufferSize = 65536;
+ @NotEmpty(message = "SSL 私钥路径不能为空")
+ private String sslKeyPath;
/**
- * 会话超时时间(毫秒,默认 60 秒)
+ * 密钥库(KeyStore)路径
*
- * 用于清理不活跃的设备地址映射
- */
- private Long sessionTimeoutMs = 60000L;
-
- /**
- * 会话清理间隔(毫秒,默认 30 秒)
- */
- private Long sessionCleanIntervalMs = 30000L;
-
- }
-
- @Data
- public static class CoapProperties {
-
- /**
- * 是否开启
- */
- @NotNull(message = "是否开启不能为空")
- private Boolean enabled;
-
- /**
- * 服务端口(CoAP 默认端口 5683)
- */
- @NotNull(message = "服务端口不能为空")
- private Integer port = 5683;
-
- /**
- * 最大消息大小(字节)
- */
- @NotNull(message = "最大消息大小不能为空")
- private Integer maxMessageSize = 1024;
-
- /**
- * ACK 超时时间(毫秒)
- */
- @NotNull(message = "ACK 超时时间不能为空")
- private Integer ackTimeout = 2000;
-
- /**
- * 最大重传次数
- */
- @NotNull(message = "最大重传次数不能为空")
- private Integer maxRetransmit = 4;
-
- }
-
- @Data
- public static class WebSocketProperties {
-
- /**
- * 是否开启
- */
- @NotNull(message = "是否开启不能为空")
- private Boolean enabled;
-
- /**
- * 服务器端口(默认:8094)
- */
- private Integer port = 8094;
-
- /**
- * WebSocket 路径(默认:/ws)
- */
- @NotEmpty(message = "WebSocket 路径不能为空")
- private String path = "/ws";
-
- /**
- * 最大消息大小(字节,默认 64KB)
- */
- private Integer maxMessageSize = 65536;
-
- /**
- * 最大帧大小(字节,默认 64KB)
+ * 包含客户端自己的证书和私钥,用于向服务端证明身份(双向认证)
*/
- private Integer maxFrameSize = 65536;
-
+ private String keyStorePath;
/**
- * 空闲超时时间(秒,默认 60)
+ * 密钥库密码
*/
- private Integer idleTimeoutSeconds = 60;
+ private String keyStorePassword;
/**
- * 是否启用 SSL(wss://)
- */
- private Boolean sslEnabled = false;
-
- /**
- * SSL 证书路径
+ * 信任库(TrustStore)路径
+ *
+ * 包含服务端信任的 CA 证书,用于验证服务端的身份
*/
- private String sslCertPath;
-
+ private String trustStorePath;
/**
- * SSL 私钥路径
+ * 信任库密码
*/
- private String sslKeyPath;
+ private String trustStorePassword;
}
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/IotProtocol.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/IotProtocol.java
new file mode 100644
index 0000000..bdfc28b
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/IotProtocol.java
@@ -0,0 +1,52 @@
+package cn.iocoder.yudao.module.iot.gateway.protocol;
+
+import cn.iocoder.yudao.module.iot.core.enums.IotProtocolTypeEnum;
+
+/**
+ * IoT 协议接口
+ *
+ * 定义传输层协议的生命周期管理
+ *
+ * @author 芋道源码
+ */
+public interface IotProtocol {
+
+ /**
+ * 获取协议实例 ID
+ *
+ * @return 协议实例 ID,如 "http-alink"、"tcp-binary"
+ */
+ String getId();
+
+ /**
+ * 获取服务器 ID(用于消息追踪,全局唯一)
+ *
+ * @return 服务器 ID
+ */
+ String getServerId();
+
+ /**
+ * 获取协议类型
+ *
+ * @return 协议类型枚举
+ */
+ IotProtocolTypeEnum getType();
+
+ /**
+ * 启动协议服务
+ */
+ void start();
+
+ /**
+ * 停止协议服务
+ */
+ void stop();
+
+ /**
+ * 检查协议服务是否正在运行
+ *
+ * @return 是否正在运行
+ */
+ boolean isRunning();
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/IotEmqxDownstreamSubscriber.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/IotProtocolDownstreamSubscriber.java
index 61bf123..2e2150f 100644
--- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/IotEmqxDownstreamSubscriber.java
+++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/IotProtocolDownstreamSubscriber.java
@@ -1,68 +1,79 @@
-package cn.iocoder.yudao.module.iot.gateway.protocol.emqx;
+package cn.iocoder.yudao.module.iot.gateway.protocol;
+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.emqx.router.IotEmqxDownstreamHandler;
-import jakarta.annotation.PostConstruct;
+import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
/**
- * IoT 网关 EMQX 订阅者:接收下行给设备的消息
+ * IoT 协议下行消息订阅者抽象类
+ *
+ * 负责接收来自消息总线的下行消息,并委托给子类进行业务处理
*
* @author 芋道源码
*/
+@AllArgsConstructor
@Slf4j
-public class IotEmqxDownstreamSubscriber implements IotMessageSubscriber {
+public abstract class IotProtocolDownstreamSubscriber implements IotMessageSubscriber {
- private final IotEmqxDownstreamHandler downstreamHandler;
+ private final IotProtocol protocol;
private final IotMessageBus messageBus;
- private final IotEmqxUpstreamProtocol protocol;
-
- public IotEmqxDownstreamSubscriber(IotEmqxUpstreamProtocol protocol, IotMessageBus messageBus) {
- this.protocol = protocol;
- this.messageBus = messageBus;
- this.downstreamHandler = new IotEmqxDownstreamHandler(protocol);
- }
-
- @PostConstruct
- public void init() {
- messageBus.register(this);
- }
-
@Override
public String getTopic() {
return IotDeviceMessageUtils.buildMessageBusGatewayDeviceMessageTopic(protocol.getServerId());
}
+ /**
+ * 保证点对点消费,需要保证独立的 Group,所以使用 Topic 作为 Group
+ */
@Override
public String getGroup() {
- // 保证点对点消费,需要保证独立的 Group,所以使用 Topic 作为 Group
return getTopic();
}
@Override
+ public void start() {
+ messageBus.register(this);
+ log.info("[start][{} 下行消息订阅成功,Topic:{}]", protocol.getType().name(), getTopic());
+ }
+
+ @Override
+ public void stop() {
+ messageBus.unregister(this);
+ log.info("[stop][{} 下行消息订阅已停止,Topic:{}]", protocol.getType().name(), getTopic());
+ }
+
+ @Override
public void onMessage(IotDeviceMessage message) {
log.debug("[onMessage][接收到下行消息, messageId: {}, method: {}, deviceId: {}]",
message.getId(), message.getMethod(), message.getDeviceId());
try {
// 1. 校验
String method = message.getMethod();
- if (method == null) {
+ if (StrUtil.isBlank(method)) {
log.warn("[onMessage][消息方法为空, messageId: {}, deviceId: {}]",
message.getId(), message.getDeviceId());
return;
}
// 2. 处理下行消息
- downstreamHandler.handle(message);
+ handleMessage(message);
} catch (Exception e) {
log.error("[onMessage][处理下行消息失败, messageId: {}, method: {}, deviceId: {}]",
message.getId(), message.getMethod(), message.getDeviceId(), e);
}
}
-}
\ No newline at end of file
+ /**
+ * 处理下行消息
+ *
+ * @param message 下行消息
+ */
+ protected abstract void handleMessage(IotDeviceMessage message);
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/IotProtocolManager.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/IotProtocolManager.java
new file mode 100644
index 0000000..3cd00c7
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/IotProtocolManager.java
@@ -0,0 +1,191 @@
+package cn.iocoder.yudao.module.iot.gateway.protocol;
+
+import cn.hutool.core.collection.CollUtil;
+import cn.hutool.core.util.BooleanUtil;
+import cn.iocoder.yudao.module.iot.core.enums.IotProtocolTypeEnum;
+import cn.iocoder.yudao.module.iot.gateway.config.IotGatewayProperties;
+import cn.iocoder.yudao.module.iot.gateway.protocol.coap.IotCoapProtocol;
+import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.IotEmqxProtocol;
+import cn.iocoder.yudao.module.iot.gateway.protocol.http.IotHttpProtocol;
+import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttProtocol;
+import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.IotTcpProtocol;
+import cn.iocoder.yudao.module.iot.gateway.protocol.udp.IotUdpProtocol;
+import cn.iocoder.yudao.module.iot.gateway.protocol.websocket.IotWebSocketProtocol;
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.SmartLifecycle;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * IoT 协议管理器:负责根据配置创建和管理协议实例
+ *
+ * @author 芋道源码
+ */
+@Slf4j
+public class IotProtocolManager implements SmartLifecycle {
+
+ private final IotGatewayProperties gatewayProperties;
+
+ /**
+ * 协议实例列表
+ */
+ private final List protocols = new ArrayList<>();
+
+ @Getter
+ private volatile boolean running = false;
+
+ public IotProtocolManager(IotGatewayProperties gatewayProperties) {
+ this.gatewayProperties = gatewayProperties;
+ }
+
+ @Override
+ public void start() {
+ if (running) {
+ return;
+ }
+ List protocolConfigs = gatewayProperties.getProtocols();
+ if (CollUtil.isEmpty(protocolConfigs)) {
+ log.info("[start][没有配置协议实例,跳过启动]");
+ return;
+ }
+
+ for (IotGatewayProperties.ProtocolProperties config : protocolConfigs) {
+ if (BooleanUtil.isFalse(config.getEnabled())) {
+ log.info("[start][协议实例 {} 未启用,跳过]", config.getId());
+ continue;
+ }
+ IotProtocol protocol = createProtocol(config);
+ if (protocol == null) {
+ continue;
+ }
+ protocol.start();
+ protocols.add(protocol);
+ }
+ running = true;
+ log.info("[start][协议管理器启动完成,共启动 {} 个协议实例]", protocols.size());
+ }
+
+ @Override
+ public void stop() {
+ if (!running) {
+ return;
+ }
+ for (IotProtocol protocol : protocols) {
+ try {
+ protocol.stop();
+ } catch (Exception e) {
+ log.error("[stop][协议实例 {} 停止失败]", protocol.getId(), e);
+ }
+ }
+ protocols.clear();
+ running = false;
+ log.info("[stop][协议管理器已停止]");
+ }
+
+ /**
+ * 创建协议实例
+ *
+ * @param config 协议实例配置
+ * @return 协议实例
+ */
+ @SuppressWarnings({"EnhancedSwitchMigration"})
+ private IotProtocol createProtocol(IotGatewayProperties.ProtocolProperties config) {
+ IotProtocolTypeEnum protocolType = IotProtocolTypeEnum.of(config.getProtocol());
+ if (protocolType == null) {
+ log.error("[createProtocol][协议实例 {} 的协议类型 {} 不存在]", config.getId(), config.getProtocol());
+ return null;
+ }
+ switch (protocolType) {
+ case HTTP:
+ return createHttpProtocol(config);
+ case TCP:
+ return createTcpProtocol(config);
+ case UDP:
+ return createUdpProtocol(config);
+ case COAP:
+ return createCoapProtocol(config);
+ case WEBSOCKET:
+ return createWebSocketProtocol(config);
+ case MQTT:
+ return createMqttProtocol(config);
+ case EMQX:
+ return createEmqxProtocol(config);
+ default:
+ throw new IllegalArgumentException(String.format(
+ "[createProtocol][协议实例 %s 的协议类型 %s 暂不支持]", config.getId(), protocolType));
+ }
+ }
+
+ /**
+ * 创建 HTTP 协议实例
+ *
+ * @param config 协议实例配置
+ * @return HTTP 协议实例
+ */
+ private IotHttpProtocol createHttpProtocol(IotGatewayProperties.ProtocolProperties config) {
+ return new IotHttpProtocol(config);
+ }
+
+ /**
+ * 创建 TCP 协议实例
+ *
+ * @param config 协议实例配置
+ * @return TCP 协议实例
+ */
+ private IotTcpProtocol createTcpProtocol(IotGatewayProperties.ProtocolProperties config) {
+ return new IotTcpProtocol(config);
+ }
+
+ /**
+ * 创建 UDP 协议实例
+ *
+ * @param config 协议实例配置
+ * @return UDP 协议实例
+ */
+ private IotUdpProtocol createUdpProtocol(IotGatewayProperties.ProtocolProperties config) {
+ return new IotUdpProtocol(config);
+ }
+
+ /**
+ * 创建 CoAP 协议实例
+ *
+ * @param config 协议实例配置
+ * @return CoAP 协议实例
+ */
+ private IotCoapProtocol createCoapProtocol(IotGatewayProperties.ProtocolProperties config) {
+ return new IotCoapProtocol(config);
+ }
+
+ /**
+ * 创建 WebSocket 协议实例
+ *
+ * @param config 协议实例配置
+ * @return WebSocket 协议实例
+ */
+ private IotWebSocketProtocol createWebSocketProtocol(IotGatewayProperties.ProtocolProperties config) {
+ return new IotWebSocketProtocol(config);
+ }
+
+ /**
+ * 创建 MQTT 协议实例
+ *
+ * @param config 协议实例配置
+ * @return MQTT 协议实例
+ */
+ private IotMqttProtocol createMqttProtocol(IotGatewayProperties.ProtocolProperties config) {
+ return new IotMqttProtocol(config);
+ }
+
+ /**
+ * 创建 EMQX 协议实例
+ *
+ * @param config 协议实例配置
+ * @return EMQX 协议实例
+ */
+ private IotEmqxProtocol createEmqxProtocol(IotGatewayProperties.ProtocolProperties config) {
+ return new IotEmqxProtocol(config);
+ }
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/IotCoapConfig.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/IotCoapConfig.java
new file mode 100644
index 0000000..45fe300
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/IotCoapConfig.java
@@ -0,0 +1,36 @@
+package cn.iocoder.yudao.module.iot.gateway.protocol.coap;
+
+import jakarta.validation.constraints.Min;
+import jakarta.validation.constraints.NotNull;
+import lombok.Data;
+
+/**
+ * IoT CoAP 协议配置
+ *
+ * @author 芋道源码
+ */
+@Data
+public class IotCoapConfig {
+
+ /**
+ * 最大消息大小(字节)
+ */
+ @NotNull(message = "最大消息大小不能为空")
+ @Min(value = 64, message = "最大消息大小必须大于 64 字节")
+ private Integer maxMessageSize = 1024;
+
+ /**
+ * ACK 超时时间(毫秒)
+ */
+ @NotNull(message = "ACK 超时时间不能为空")
+ @Min(value = 100, message = "ACK 超时时间必须大于 100 毫秒")
+ private Integer ackTimeoutMs = 2000;
+
+ /**
+ * 最大重传次数
+ */
+ @NotNull(message = "最大重传次数不能为空")
+ @Min(value = 0, message = "最大重传次数必须大于等于 0")
+ private Integer maxRetransmit = 4;
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/IotCoapDownstreamSubscriber.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/IotCoapDownstreamSubscriber.java
deleted file mode 100644
index d01cdc4..0000000
--- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/IotCoapDownstreamSubscriber.java
+++ /dev/null
@@ -1,46 +0,0 @@
-package cn.iocoder.yudao.module.iot.gateway.protocol.coap;
-
-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 jakarta.annotation.PostConstruct;
-import lombok.RequiredArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
-
-/**
- * IoT 网关 CoAP 订阅者:接收下行给设备的消息
- *
- * @author 芋道源码
- */
-@RequiredArgsConstructor
-@Slf4j
-public class IotCoapDownstreamSubscriber implements IotMessageSubscriber {
-
- private final IotCoapUpstreamProtocol protocol;
-
- private final IotMessageBus messageBus;
-
- @PostConstruct
- public void init() {
- messageBus.register(this);
- }
-
- @Override
- public String getTopic() {
- return IotDeviceMessageUtils.buildMessageBusGatewayDeviceMessageTopic(protocol.getServerId());
- }
-
- @Override
- public String getGroup() {
- // 保证点对点消费,需要保证独立的 Group,所以使用 Topic 作为 Group
- return getTopic();
- }
-
- @Override
- public void onMessage(IotDeviceMessage message) {
- // 如需支持,可通过 CoAP Observe 模式实现(设备订阅资源,服务器推送变更)
- log.warn("[onMessage][IoT 网关 CoAP 协议暂不支持下行消息,忽略消息:{}]", message);
- }
-
-}
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/IotCoapProtocol.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/IotCoapProtocol.java
new file mode 100644
index 0000000..14fe10d
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/IotCoapProtocol.java
@@ -0,0 +1,173 @@
+package cn.iocoder.yudao.module.iot.gateway.protocol.coap;
+
+import cn.hutool.extra.spring.SpringUtil;
+import cn.iocoder.yudao.module.iot.core.enums.IotProtocolTypeEnum;
+import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
+import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
+import cn.iocoder.yudao.module.iot.gateway.config.IotGatewayProperties.ProtocolProperties;
+import cn.iocoder.yudao.module.iot.gateway.protocol.IotProtocol;
+import cn.iocoder.yudao.module.iot.gateway.protocol.coap.handler.downstream.IotCoapDownstreamSubscriber;
+import cn.iocoder.yudao.module.iot.gateway.protocol.coap.handler.upstream.IotCoapAuthHandler;
+import cn.iocoder.yudao.module.iot.gateway.protocol.coap.handler.upstream.IotCoapAuthResource;
+import cn.iocoder.yudao.module.iot.gateway.protocol.coap.handler.upstream.IotCoapRegisterHandler;
+import cn.iocoder.yudao.module.iot.gateway.protocol.coap.handler.upstream.IotCoapRegisterResource;
+import cn.iocoder.yudao.module.iot.gateway.protocol.coap.handler.upstream.IotCoapRegisterSubHandler;
+import cn.iocoder.yudao.module.iot.gateway.protocol.coap.handler.upstream.IotCoapRegisterSubResource;
+import cn.iocoder.yudao.module.iot.gateway.protocol.coap.handler.upstream.IotCoapUpstreamHandler;
+import cn.iocoder.yudao.module.iot.gateway.protocol.coap.handler.upstream.IotCoapUpstreamTopicResource;
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.californium.core.CoapResource;
+import org.eclipse.californium.core.CoapServer;
+import org.eclipse.californium.core.config.CoapConfig;
+import org.eclipse.californium.elements.config.Configuration;
+import cn.hutool.core.lang.Assert;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * IoT CoAP 协议实现
+ *
+ * 基于 Eclipse Californium 实现,支持:
+ * 1. 认证:POST /auth
+ * 2. 设备动态注册:POST /auth/register/device
+ * 3. 子设备动态注册:POST /auth/register/sub-device/{productKey}/{deviceName}
+ * 4. 属性上报:POST /topic/sys/{productKey}/{deviceName}/thing/property/post
+ * 5. 事件上报:POST /topic/sys/{productKey}/{deviceName}/thing/event/post
+ *
+ * @author 芋道源码
+ */
+@Slf4j
+public class IotCoapProtocol implements IotProtocol {
+
+ /**
+ * 协议配置
+ */
+ private final ProtocolProperties properties;
+ /**
+ * 服务器 ID(用于消息追踪,全局唯一)
+ */
+ @Getter
+ private final String serverId;
+
+ /**
+ * 运行状态
+ */
+ @Getter
+ private volatile boolean running = false;
+
+ /**
+ * CoAP 服务器
+ */
+ private CoapServer coapServer;
+
+ /**
+ * 下行消息订阅者
+ */
+ private final IotCoapDownstreamSubscriber downstreamSubscriber;
+
+ public IotCoapProtocol(ProtocolProperties properties) {
+ IotCoapConfig coapConfig = properties.getCoap();
+ Assert.notNull(coapConfig, "CoAP 协议配置(coap)不能为空");
+ this.properties = properties;
+ this.serverId = IotDeviceMessageUtils.generateServerId(properties.getPort());
+
+ // 初始化下行消息订阅者
+ IotMessageBus messageBus = SpringUtil.getBean(IotMessageBus.class);
+ this.downstreamSubscriber = new IotCoapDownstreamSubscriber(this, messageBus);
+ }
+
+ @Override
+ public String getId() {
+ return properties.getId();
+ }
+
+ @Override
+ public IotProtocolTypeEnum getType() {
+ return IotProtocolTypeEnum.COAP;
+ }
+
+ @Override
+ public void start() {
+ if (running) {
+ log.warn("[start][IoT CoAP 协议 {} 已经在运行中]", getId());
+ return;
+ }
+
+ IotCoapConfig coapConfig = properties.getCoap();
+ try {
+ // 1.1 创建 CoAP 配置
+ Configuration config = Configuration.createStandardWithoutFile();
+ config.set(CoapConfig.COAP_PORT, properties.getPort());
+ config.set(CoapConfig.MAX_MESSAGE_SIZE, coapConfig.getMaxMessageSize());
+ config.set(CoapConfig.ACK_TIMEOUT, coapConfig.getAckTimeoutMs(), TimeUnit.MILLISECONDS);
+ config.set(CoapConfig.MAX_RETRANSMIT, coapConfig.getMaxRetransmit());
+ // 1.2 创建 CoAP 服务器
+ coapServer = new CoapServer(config);
+
+ // 2.1 添加 /auth 认证资源
+ IotCoapAuthHandler authHandler = new IotCoapAuthHandler(serverId);
+ IotCoapAuthResource authResource = new IotCoapAuthResource(authHandler);
+ coapServer.add(authResource);
+ // 2.2 添加 /auth/register/device 设备动态注册资源(一型一密)
+ IotCoapRegisterHandler registerHandler = new IotCoapRegisterHandler();
+ IotCoapRegisterResource registerResource = new IotCoapRegisterResource(registerHandler);
+ // 2.3 添加 /auth/register/sub-device/{productKey}/{deviceName} 子设备动态注册资源
+ IotCoapRegisterSubHandler registerSubHandler = new IotCoapRegisterSubHandler();
+ IotCoapRegisterSubResource registerSubResource = new IotCoapRegisterSubResource(registerSubHandler);
+ authResource.add(new CoapResource("register") {{
+ add(registerResource);
+ add(registerSubResource);
+ }});
+ // 2.4 添加 /topic 根资源(用于上行消息)
+ IotCoapUpstreamHandler upstreamHandler = new IotCoapUpstreamHandler(serverId);
+ IotCoapUpstreamTopicResource topicResource = new IotCoapUpstreamTopicResource(serverId, upstreamHandler);
+ coapServer.add(topicResource);
+
+ // 3. 启动服务器
+ coapServer.start();
+ running = true;
+ log.info("[start][IoT CoAP 协议 {} 启动成功,端口:{},serverId:{}]",
+ getId(), properties.getPort(), serverId);
+
+ // 4. 启动下行消息订阅者
+ this.downstreamSubscriber.start();
+ } catch (Exception e) {
+ log.error("[start][IoT CoAP 协议 {} 启动失败]", getId(), e);
+ if (coapServer != null) {
+ coapServer.destroy();
+ coapServer = null;
+ }
+ throw e;
+ }
+ }
+
+ @Override
+ public void stop() {
+ if (!running) {
+ return;
+ }
+ // 1. 停止下行消息订阅者
+ try {
+ downstreamSubscriber.stop();
+ log.info("[stop][IoT CoAP 协议 {} 下行消息订阅者已停止]", getId());
+ } catch (Exception e) {
+ log.error("[stop][IoT CoAP 协议 {} 下行消息订阅者停止失败]", getId(), e);
+ }
+
+ // 2. 关闭 CoAP 服务器
+ if (coapServer != null) {
+ try {
+ coapServer.stop();
+ coapServer.destroy();
+ coapServer = null;
+ log.info("[stop][IoT CoAP 协议 {} 服务器已停止]", getId());
+ } catch (Exception e) {
+ log.error("[stop][IoT CoAP 协议 {} 服务器停止失败]", getId(), e);
+ }
+ }
+ running = false;
+ log.info("[stop][IoT CoAP 协议 {} 已停止]", getId());
+ }
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/IotCoapUpstreamProtocol.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/IotCoapUpstreamProtocol.java
deleted file mode 100644
index e259aa6..0000000
--- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/IotCoapUpstreamProtocol.java
+++ /dev/null
@@ -1,95 +0,0 @@
-package cn.iocoder.yudao.module.iot.gateway.protocol.coap;
-
-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.coap.router.IotCoapAuthHandler;
-import cn.iocoder.yudao.module.iot.gateway.protocol.coap.router.IotCoapAuthResource;
-import cn.iocoder.yudao.module.iot.gateway.protocol.coap.router.IotCoapRegisterHandler;
-import cn.iocoder.yudao.module.iot.gateway.protocol.coap.router.IotCoapRegisterResource;
-import cn.iocoder.yudao.module.iot.gateway.protocol.coap.router.IotCoapUpstreamTopicResource;
-import cn.iocoder.yudao.module.iot.gateway.protocol.coap.router.IotCoapUpstreamHandler;
-import jakarta.annotation.PostConstruct;
-import jakarta.annotation.PreDestroy;
-import lombok.Getter;
-import lombok.extern.slf4j.Slf4j;
-import org.eclipse.californium.core.CoapResource;
-import org.eclipse.californium.core.CoapServer;
-import org.eclipse.californium.core.config.CoapConfig;
-import org.eclipse.californium.elements.config.Configuration;
-
-import java.util.concurrent.TimeUnit;
-
-/**
- * IoT 网关 CoAP 协议:接收设备上行消息
- *
- * 基于 Eclipse Californium 实现,支持:
- * 1. 认证:POST /auth
- * 2. 属性上报:POST /topic/sys/{productKey}/{deviceName}/thing/property/post
- * 3. 事件上报:POST /topic/sys/{productKey}/{deviceName}/thing/event/post
- *
- * @author 芋道源码
- */
-@Slf4j
-public class IotCoapUpstreamProtocol {
-
- private final IotGatewayProperties.CoapProperties coapProperties;
-
- private CoapServer coapServer;
-
- @Getter
- private final String serverId;
-
- public IotCoapUpstreamProtocol(IotGatewayProperties.CoapProperties coapProperties) {
- this.coapProperties = coapProperties;
- this.serverId = IotDeviceMessageUtils.generateServerId(coapProperties.getPort());
- }
-
- @PostConstruct
- public void start() {
- try {
- // 1.1 创建网络配置(Californium 3.x API)
- Configuration config = Configuration.createStandardWithoutFile();
- config.set(CoapConfig.COAP_PORT, coapProperties.getPort());
- config.set(CoapConfig.MAX_MESSAGE_SIZE, coapProperties.getMaxMessageSize());
- config.set(CoapConfig.ACK_TIMEOUT, coapProperties.getAckTimeout(), TimeUnit.MILLISECONDS);
- config.set(CoapConfig.MAX_RETRANSMIT, coapProperties.getMaxRetransmit());
- // 1.2 创建 CoAP 服务器
- coapServer = new CoapServer(config);
-
- // 2.1 添加 /auth 认证资源
- IotCoapAuthHandler authHandler = new IotCoapAuthHandler();
- IotCoapAuthResource authResource = new IotCoapAuthResource(this, authHandler);
- coapServer.add(authResource);
- // 2.2 添加 /auth/register/device 设备动态注册资源(一型一密)
- IotCoapRegisterHandler registerHandler = new IotCoapRegisterHandler();
- IotCoapRegisterResource registerResource = new IotCoapRegisterResource(registerHandler);
- authResource.add(new CoapResource("register") {{
- add(registerResource);
- }});
- // 2.3 添加 /topic 根资源(用于上行消息)
- IotCoapUpstreamHandler upstreamHandler = new IotCoapUpstreamHandler();
- IotCoapUpstreamTopicResource topicResource = new IotCoapUpstreamTopicResource(this, upstreamHandler);
- coapServer.add(topicResource);
-
- // 3. 启动服务器
- coapServer.start();
- log.info("[start][IoT 网关 CoAP 协议启动成功,端口:{},资源:/auth, /auth/register/device, /topic]", coapProperties.getPort());
- } catch (Exception e) {
- log.error("[start][IoT 网关 CoAP 协议启动失败]", e);
- throw e;
- }
- }
-
- @PreDestroy
- public void stop() {
- if (coapServer != null) {
- try {
- coapServer.stop();
- log.info("[stop][IoT 网关 CoAP 协议已停止]");
- } catch (Exception e) {
- log.error("[stop][IoT 网关 CoAP 协议停止失败]", e);
- }
- }
- }
-
-}
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/handler/downstream/IotCoapDownstreamSubscriber.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/handler/downstream/IotCoapDownstreamSubscriber.java
new file mode 100644
index 0000000..188d2e6
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/handler/downstream/IotCoapDownstreamSubscriber.java
@@ -0,0 +1,27 @@
+package cn.iocoder.yudao.module.iot.gateway.protocol.coap.handler.downstream;
+
+import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
+import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
+import cn.iocoder.yudao.module.iot.gateway.protocol.IotProtocolDownstreamSubscriber;
+import cn.iocoder.yudao.module.iot.gateway.protocol.coap.IotCoapProtocol;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * IoT 网关 CoAP 订阅者:接收下行给设备的消息
+ *
+ * @author 芋道源码
+ */
+@Slf4j
+public class IotCoapDownstreamSubscriber extends IotProtocolDownstreamSubscriber {
+
+ public IotCoapDownstreamSubscriber(IotCoapProtocol protocol, IotMessageBus messageBus) {
+ super(protocol, messageBus);
+ }
+
+ @Override
+ protected void handleMessage(IotDeviceMessage message) {
+ // 如需支持,可通过 CoAP Observe 模式实现(设备订阅资源,服务器推送变更)
+ log.warn("[handleMessage][IoT 网关 CoAP 协议暂不支持下行消息,忽略消息:{}]", message);
+ }
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/handler/upstream/IotCoapAbstractHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/handler/upstream/IotCoapAbstractHandler.java
new file mode 100644
index 0000000..994fb14
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/handler/upstream/IotCoapAbstractHandler.java
@@ -0,0 +1,186 @@
+package cn.iocoder.yudao.module.iot.gateway.protocol.coap.handler.upstream;
+
+import cn.hutool.core.collection.CollUtil;
+import cn.hutool.core.util.ArrayUtil;
+import cn.hutool.core.util.ObjUtil;
+import cn.hutool.core.util.StrUtil;
+import cn.hutool.extra.spring.SpringUtil;
+import cn.iocoder.yudao.framework.common.exception.ServiceException;
+import cn.iocoder.yudao.framework.common.pojo.CommonResult;
+import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
+import cn.iocoder.yudao.module.iot.core.topic.IotDeviceIdentity;
+import cn.iocoder.yudao.module.iot.gateway.service.auth.IotDeviceTokenService;
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.californium.core.coap.CoAP;
+import org.eclipse.californium.core.coap.MediaTypeRegistry;
+import org.eclipse.californium.core.coap.Option;
+import org.eclipse.californium.core.server.resources.CoapExchange;
+
+import java.util.List;
+
+import static cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants.*;
+import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
+
+/**
+ * IoT 网关 CoAP 协议的处理器抽象基类:提供通用的前置处理(认证)、请求解析、响应处理、全局的异常捕获等
+ *
+ * @author 芋道源码
+ */
+@Slf4j
+public abstract class IotCoapAbstractHandler {
+
+ /**
+ * 自定义 CoAP Option 编号,用于携带 Token
+ *
+ * CoAP Option 范围 2048-65535 属于实验/自定义范围
+ */
+ public static final int OPTION_TOKEN = 2088;
+
+ private final IotDeviceTokenService deviceTokenService = SpringUtil.getBean(IotDeviceTokenService.class);
+
+ /**
+ * 处理 CoAP 请求(模板方法)
+ *
+ * @param exchange CoAP 交换对象
+ */
+ public final void handle(CoapExchange exchange) {
+ try {
+ // 1. 前置处理
+ beforeHandle(exchange);
+
+ // 2. 执行业务逻辑
+ CommonResult