BJ_ZTEITS-Developers / parking-cloud-yudao

Commit bec8cc6ef8a3a604f742664bc2368e3caf4f5d96

Authored by YunaiV
1 parent 3027caa1

feat：【iot】coap 协议接入 50%：初始化整体实现，基于 pure-wishing-muffin.md 规划

Inline Side-by-side
Showing 14 changed files with 877 additions and 0 deletions
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayConfiguration.java
  View file @bec8cc6
1 package cn.iocoder.yudao.module.iot.gateway.config; 1 package cn.iocoder.yudao.module.iot.gateway.config;
2 2
3 import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus; 3 import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
  4 +import cn.iocoder.yudao.module.iot.gateway.protocol.coap.IotCoapDownstreamSubscriber;
  5 +import cn.iocoder.yudao.module.iot.gateway.protocol.coap.IotCoapUpstreamProtocol;
  6 +import cn.iocoder.yudao.module.iot.gateway.protocol.coap.router.IotCoapAuthHandler;
  7 +import cn.iocoder.yudao.module.iot.gateway.protocol.coap.router.IotCoapUpstreamHandler;
4 import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.IotEmqxAuthEventProtocol; 8 import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.IotEmqxAuthEventProtocol;
5 import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.IotEmqxDownstreamSubscriber; 9 import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.IotEmqxDownstreamSubscriber;
6 import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.IotEmqxUpstreamProtocol; 10 import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.IotEmqxUpstreamProtocol;
@@ -194,4 +198,28 @@ public class IotGatewayConfiguration { @@ -194,4 +198,28 @@ public class IotGatewayConfiguration {
194 198
195 } 199 }
196 200
  201 + /**
  202 + * IoT 网关 CoAP 协议配置类
  203 + */
  204 + @Configuration
  205 + @ConditionalOnProperty(prefix = "yudao.iot.gateway.protocol.coap", name = "enabled", havingValue = "true")
  206 + @Slf4j
  207 + public static class CoapProtocolConfiguration {
  208 +
  209 + @Bean
  210 + public IotCoapUpstreamProtocol iotCoapUpstreamProtocol(IotGatewayProperties gatewayProperties,
  211 + IotCoapAuthHandler authHandler,
  212 + IotCoapUpstreamHandler upstreamHandler) {
  213 + return new IotCoapUpstreamProtocol(gatewayProperties.getProtocol().getCoap(),
  214 + authHandler, upstreamHandler);
  215 + }
  216 +
  217 + @Bean
  218 + public IotCoapDownstreamSubscriber iotCoapDownstreamSubscriber(IotCoapUpstreamProtocol coapUpstreamProtocol,
  219 + IotMessageBus messageBus) {
  220 + return new IotCoapDownstreamSubscriber(coapUpstreamProtocol, messageBus);
  221 + }
  222 +
  223 + }
  224 +
197 } 225 }
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayProperties.java
  View file @bec8cc6
@@ -93,6 +93,11 @@ public class IotGatewayProperties { @@ -93,6 +93,11 @@ public class IotGatewayProperties {
93 */ 93 */
94 private MqttWsProperties mqttWs; 94 private MqttWsProperties mqttWs;
95 95
  96 + /**
  97 + * CoAP 组件配置
  98 + */
  99 + private CoapProperties coap;
  100 +
96 } 101 }
97 102
98 @Data 103 @Data
@@ -503,4 +508,39 @@ public class IotGatewayProperties { @@ -503,4 +508,39 @@ public class IotGatewayProperties {
503 508
504 } 509 }
505 510
  511 + @Data
  512 + public static class CoapProperties {
  513 +
  514 + /**
  515 + * 是否开启
  516 + */
  517 + @NotNull(message = "是否开启不能为空")
  518 + private Boolean enabled;
  519 +
  520 + /**
  521 + * 服务端口（CoAP 默认端口 5683）
  522 + */
  523 + // TODO @AI：默认不为空
  524 + private Integer port = 5683;
  525 +
  526 + /**
  527 + * 最大消息大小（字节）
  528 + */
  529 + // TODO @AI：默认不为空
  530 + private Integer maxMessageSize = 1024;
  531 +
  532 + /**
  533 + * ACK 超时时间（毫秒）
  534 + */
  535 + // TODO @AI：默认不为空
  536 + private Integer ackTimeout = 2000;
  537 +
  538 + /**
  539 + * 最大重传次数
  540 + */
  541 + // TODO @AI：默认不为空
  542 + private Integer maxRetransmit = 4;
  543 +
  544 + }
  545 +
506 } 546 }
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/IotCoapDownstreamSubscriber.java 0 → 100644
  View file @bec8cc6
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.coap;
  2 +
  3 +import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
  4 +import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber;
  5 +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
  6 +import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
  7 +import jakarta.annotation.PostConstruct;
  8 +import lombok.RequiredArgsConstructor;
  9 +import lombok.extern.slf4j.Slf4j;
  10 +
  11 +/**
  12 + * IoT 网关 CoAP 订阅者：接收下行给设备的消息
  13 + *
  14 + * @author 芋道源码
  15 + */
  16 +@RequiredArgsConstructor
  17 +@Slf4j
  18 +public class IotCoapDownstreamSubscriber implements IotMessageSubscriber<IotDeviceMessage> {
  19 +
  20 + private final IotCoapUpstreamProtocol protocol;
  21 +
  22 + private final IotMessageBus messageBus;
  23 +
  24 + @PostConstruct
  25 + public void init() {
  26 + messageBus.register(this);
  27 + }
  28 +
  29 + @Override
  30 + public String getTopic() {
  31 + return IotDeviceMessageUtils.buildMessageBusGatewayDeviceMessageTopic(protocol.getServerId());
  32 + }
  33 +
  34 + @Override
  35 + public String getGroup() {
  36 + // 保证点对点消费，需要保证独立的 Group，所以使用 Topic 作为 Group
  37 + return getTopic();
  38 + }
  39 +
  40 + @Override
  41 + public void onMessage(IotDeviceMessage message) {
  42 + // 如需支持，可通过 CoAP Observe 模式实现（设备订阅资源，服务器推送变更）
  43 + log.warn("[onMessage][IoT 网关 CoAP 协议暂不支持下行消息，忽略消息：{}]", message);
  44 + }
  45 +
  46 +}
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/IotCoapUpstreamProtocol.java 0 → 100644
  View file @bec8cc6
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.coap;
  2 +
  3 +import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
  4 +import cn.iocoder.yudao.module.iot.gateway.config.IotGatewayProperties;
  5 +import cn.iocoder.yudao.module.iot.gateway.protocol.coap.router.IotCoapAuthHandler;
  6 +import cn.iocoder.yudao.module.iot.gateway.protocol.coap.router.IotCoapAuthResource;
  7 +import cn.iocoder.yudao.module.iot.gateway.protocol.coap.router.IotCoapUpstreamTopicResource;
  8 +import cn.iocoder.yudao.module.iot.gateway.protocol.coap.router.IotCoapUpstreamHandler;
  9 +import jakarta.annotation.PostConstruct;
  10 +import jakarta.annotation.PreDestroy;
  11 +import lombok.Getter;
  12 +import lombok.extern.slf4j.Slf4j;
  13 +import org.eclipse.californium.core.CoapServer;
  14 +import org.eclipse.californium.core.config.CoapConfig;
  15 +import org.eclipse.californium.elements.config.Configuration;
  16 +
  17 +import java.util.concurrent.TimeUnit;
  18 +
  19 +/**
  20 + * IoT 网关 CoAP 协议：接收设备上行消息
  21 + *
  22 + * 基于 Eclipse Californium 实现，支持：
  23 + * 1. 认证：POST /auth
  24 + * 2. 属性上报：POST /topic/sys/{productKey}/{deviceName}/thing/property/post
  25 + * 3. 事件上报：POST /topic/sys/{productKey}/{deviceName}/thing/event/{eventId}/post
  26 + *
  27 + * @author 芋道源码
  28 + */
  29 +@Slf4j
  30 +public class IotCoapUpstreamProtocol {
  31 +
  32 + private final IotGatewayProperties.CoapProperties coapProperties;
  33 +
  34 + private final IotCoapAuthHandler authHandler;
  35 + private final IotCoapUpstreamHandler upstreamHandler;
  36 +
  37 + private CoapServer coapServer;
  38 +
  39 + @Getter
  40 + private final String serverId;
  41 +
  42 + public IotCoapUpstreamProtocol(IotGatewayProperties.CoapProperties coapProperties,
  43 + IotCoapAuthHandler authHandler,
  44 + IotCoapUpstreamHandler upstreamHandler) {
  45 + this.coapProperties = coapProperties;
  46 + this.authHandler = authHandler;
  47 + this.upstreamHandler = upstreamHandler;
  48 + this.serverId = IotDeviceMessageUtils.generateServerId(coapProperties.getPort());
  49 + }
  50 +
  51 + @PostConstruct
  52 + public void start() {
  53 + try {
  54 + // 1.1 创建网络配置（Californium 3.x API）
  55 + Configuration config = Configuration.createStandardWithoutFile();
  56 + config.set(CoapConfig.COAP_PORT, coapProperties.getPort());
  57 + config.set(CoapConfig.MAX_MESSAGE_SIZE, coapProperties.getMaxMessageSize());
  58 + config.set(CoapConfig.ACK_TIMEOUT, coapProperties.getAckTimeout(), TimeUnit.MILLISECONDS);
  59 + config.set(CoapConfig.MAX_RETRANSMIT, coapProperties.getMaxRetransmit());
  60 + // 1.2 创建 CoAP 服务器
  61 + coapServer = new CoapServer(config);
  62 +
  63 + // 2.1 添加 /auth 认证资源
  64 + IotCoapAuthResource authResource = new IotCoapAuthResource(this, authHandler);
  65 + coapServer.add(authResource);
  66 + // 2.2 添加 /topic 根资源（用于上行消息）
  67 + IotCoapUpstreamTopicResource topicResource = new IotCoapUpstreamTopicResource(this, upstreamHandler);
  68 + coapServer.add(topicResource);
  69 +
  70 + // 3. 启动服务器
  71 + coapServer.start();
  72 + log.info("[start][IoT 网关 CoAP 协议启动成功，端口：{}，资源：/auth, /topic]", coapProperties.getPort());
  73 + } catch (Exception e) {
  74 + log.error("[start][IoT 网关 CoAP 协议启动失败]", e);
  75 + throw e;
  76 + }
  77 + }
  78 +
  79 + @PreDestroy
  80 + public void stop() {
  81 + if (coapServer != null) {
  82 + try {
  83 + coapServer.stop();
  84 + log.info("[stop][IoT 网关 CoAP 协议已停止]");
  85 + } catch (Exception e) {
  86 + log.error("[stop][IoT 网关 CoAP 协议停止失败]", e);
  87 + }
  88 + }
  89 + }
  90 +
  91 +}
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/package-info.java 0 → 100644
  View file @bec8cc6
  1 +/**
  2 + * IoT 网关 CoAP 协议
  3 + *
  4 + * 基于 Eclipse Californium 实现，支持设备通过 CoAP 协议进行：
  5 + * 1. 属性上报：POST /sys/{productKey}/{deviceName}/thing/property/post
  6 + * 2. 事件上报：POST /sys/{productKey}/{deviceName}/thing/event/{eventId}/post
  7 + *
  8 + * 认证方式：通过 URI Query 参数 token 进行认证
  9 + * 示例：coap://server:5683/sys/pk/dn/thing/property/post?token=xxx
  10 + *
  11 + * @author 芋道源码
  12 + */
  13 +// TODO @AI：参考 /Users/yunai/Java/ruoyi-vue-pro-jdk25/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/package-info.java （现在注释应该有点不太对）
  14 +package cn.iocoder.yudao.module.iot.gateway.protocol.coap;
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/router/IotCoapAuthHandler.java 0 → 100644
  View file @bec8cc6
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.coap.router;
  2 +
  3 +import cn.hutool.core.lang.Assert;
  4 +import cn.hutool.core.map.MapUtil;
  5 +import cn.hutool.core.util.BooleanUtil;
  6 +import cn.hutool.core.util.StrUtil;
  7 +import cn.iocoder.yudao.framework.common.pojo.CommonResult;
  8 +import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
  9 +import cn.iocoder.yudao.module.iot.core.biz.IotDeviceCommonApi;
  10 +import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceAuthReqDTO;
  11 +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
  12 +import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils;
  13 +import cn.iocoder.yudao.module.iot.gateway.protocol.coap.IotCoapUpstreamProtocol;
  14 +import cn.iocoder.yudao.module.iot.gateway.service.auth.IotDeviceTokenService;
  15 +import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
  16 +import lombok.RequiredArgsConstructor;
  17 +import lombok.extern.slf4j.Slf4j;
  18 +import org.eclipse.californium.core.coap.CoAP;
  19 +import org.eclipse.californium.core.coap.MediaTypeRegistry;
  20 +import org.eclipse.californium.core.server.resources.CoapExchange;
  21 +import org.springframework.stereotype.Component;
  22 +
  23 +import java.util.Map;
  24 +
  25 +import static cn.iocoder.yudao.framework.common.pojo.CommonResult.success;
  26 +
  27 +/**
  28 + * IoT 网关 CoAP 协议的【认证】处理器
  29 + *
  30 + * 参考 {@link cn.iocoder.yudao.module.iot.gateway.protocol.http.router.IotHttpAuthHandler}
  31 + *
  32 + * @author 芋道源码
  33 + */
  34 +@Component
  35 +@RequiredArgsConstructor
  36 +@Slf4j
  37 +public class IotCoapAuthHandler {
  38 +
  39 + private final IotDeviceTokenService deviceTokenService;
  40 +
  41 + private final IotDeviceCommonApi deviceApi;
  42 +
  43 + private final IotDeviceMessageService deviceMessageService;
  44 +
  45 + /**
  46 + * 处理认证请求
  47 + *
  48 + * @param exchange CoAP 交换对象
  49 + * @param protocol 协议对象
  50 + */
  51 + public void handle(CoapExchange exchange, IotCoapUpstreamProtocol protocol) {
  52 + try {
  53 + // 1.1 解析请求体
  54 + byte[] payload = exchange.getRequestPayload();
  55 + if (payload == null || payload.length == 0) {
  56 + respondError(exchange, CoAP.ResponseCode.BAD_REQUEST, "请求体不能为空");
  57 + return;
  58 + }
  59 + Map<String, Object> body;
  60 + try {
  61 + body = JsonUtils.parseObject(new String(payload), Map.class);
  62 + } catch (Exception e) {
  63 + respondError(exchange, CoAP.ResponseCode.BAD_REQUEST, "请求体 JSON 格式错误");
  64 + return;
  65 + }
  66 + // TODO @AI：通过 hutool maputil 去获取，简化下；
  67 + // 1.2 解析参数
  68 + String clientId = (String) body.get("clientId");
  69 + if (StrUtil.isEmpty(clientId)) {
  70 + respondError(exchange, CoAP.ResponseCode.BAD_REQUEST, "clientId 不能为空");
  71 + return;
  72 + }
  73 + String username = (String) body.get("username");
  74 + if (StrUtil.isEmpty(username)) {
  75 + respondError(exchange, CoAP.ResponseCode.BAD_REQUEST, "username 不能为空");
  76 + return;
  77 + }
  78 + String password = (String) body.get("password");
  79 + if (StrUtil.isEmpty(password)) {
  80 + respondError(exchange, CoAP.ResponseCode.BAD_REQUEST, "password 不能为空");
  81 + return;
  82 + }
  83 +
  84 + // 2.1 执行认证
  85 + CommonResult<Boolean> result = deviceApi.authDevice(new IotDeviceAuthReqDTO()
  86 + .setClientId(clientId).setUsername(username).setPassword(password));
  87 + if (result.isError()) {
  88 + log.warn("[handle][认证失败，clientId: {}, 错误: {}]", clientId, result.getMsg());
  89 + respondError(exchange, CoAP.ResponseCode.UNAUTHORIZED, "认证失败：" + result.getMsg());
  90 + return;
  91 + }
  92 + if (!BooleanUtil.isTrue(result.getData())) {
  93 + log.warn("[handle][认证失败，clientId: {}]", clientId);
  94 + respondError(exchange, CoAP.ResponseCode.UNAUTHORIZED, "认证失败");
  95 + return;
  96 + }
  97 + // 2.2 生成 Token
  98 + IotDeviceAuthUtils.DeviceInfo deviceInfo = deviceTokenService.parseUsername(username);
  99 + Assert.notNull(deviceInfo, "设备信息不能为空");
  100 + String token = deviceTokenService.createToken(deviceInfo.getProductKey(), deviceInfo.getDeviceName());
  101 + Assert.notBlank(token, "生成 token 不能为空");
  102 +
  103 + // 3. 执行上线
  104 + IotDeviceMessage message = IotDeviceMessage.buildStateUpdateOnline();
  105 + deviceMessageService.sendDeviceMessage(message,
  106 + deviceInfo.getProductKey(), deviceInfo.getDeviceName(), protocol.getServerId());
  107 +
  108 + // 4. 返回成功响应
  109 + log.info("[handle][认证成功，productKey: {}, deviceName: {}]",
  110 + deviceInfo.getProductKey(), deviceInfo.getDeviceName());
  111 + respondSuccess(exchange, MapUtil.of("token", token));
  112 + } catch (Exception e) {
  113 + log.error("[handle][认证处理异常]", e);
  114 + respondError(exchange, CoAP.ResponseCode.INTERNAL_SERVER_ERROR, "服务器内部错误");
  115 + }
  116 + }
  117 +
  118 + // TODO @AI：抽到 coap 的 util 里；
  119 + /**
  120 + * 返回成功响应
  121 + */
  122 + private void respondSuccess(CoapExchange exchange, Object data) {
  123 + CommonResult<Object> result = success(data);
  124 + String json = JsonUtils.toJsonString(result);
  125 + exchange.respond(CoAP.ResponseCode.CONTENT, json, MediaTypeRegistry.APPLICATION_JSON);
  126 + }
  127 +
  128 + // TODO @AI：抽到 coap 的 util 里；
  129 + /**
  130 + * 返回错误响应
  131 + */
  132 + private void respondError(CoapExchange exchange, CoAP.ResponseCode code, String message) {
  133 + CommonResult<Object> result = CommonResult.error(code.value, message);
  134 + String json = JsonUtils.toJsonString(result);
  135 + exchange.respond(code, json, MediaTypeRegistry.APPLICATION_JSON);
  136 + }
  137 +
  138 +}
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/router/IotCoapAuthResource.java 0 → 100644
  View file @bec8cc6
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.coap.router;
  2 +
  3 +import cn.iocoder.yudao.module.iot.gateway.protocol.coap.IotCoapUpstreamProtocol;
  4 +import lombok.extern.slf4j.Slf4j;
  5 +import org.eclipse.californium.core.CoapResource;
  6 +import org.eclipse.californium.core.server.resources.CoapExchange;
  7 +
  8 +/**
  9 + * IoT 网关 CoAP 协议的认证资源（/auth）
  10 + *
  11 + * 设备通过此资源进行认证，获取 Token
  12 + *
  13 + * @author 芋道源码
  14 + */
  15 +@Slf4j
  16 +public class IotCoapAuthResource extends CoapResource {
  17 +
  18 + public static final String PATH = "auth";
  19 +
  20 + private final IotCoapUpstreamProtocol protocol;
  21 + private final IotCoapAuthHandler authHandler;
  22 +
  23 + public IotCoapAuthResource(IotCoapUpstreamProtocol protocol,
  24 + IotCoapAuthHandler authHandler) {
  25 + super(PATH);
  26 + this.protocol = protocol;
  27 + this.authHandler = authHandler;
  28 + log.info("[IotCoapAuthResource][创建 CoAP 认证资源: /{}]", PATH);
  29 + }
  30 +
  31 + @Override
  32 + public void handlePOST(CoapExchange exchange) {
  33 + log.debug("[handlePOST][收到 /auth POST 请求]");
  34 + authHandler.handle(exchange, protocol);
  35 + }
  36 +
  37 +}
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/router/IotCoapUpstreamHandler.java 0 → 100644
  View file @bec8cc6
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.coap.router;
  2 +
  3 +import cn.hutool.core.map.MapUtil;
  4 +import cn.hutool.core.text.StrPool;
  5 +import cn.hutool.core.util.ObjUtil;
  6 +import cn.hutool.core.util.StrUtil;
  7 +import cn.iocoder.yudao.framework.common.pojo.CommonResult;
  8 +import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
  9 +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
  10 +import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils;
  11 +import cn.iocoder.yudao.module.iot.gateway.protocol.coap.IotCoapUpstreamProtocol;
  12 +import cn.iocoder.yudao.module.iot.gateway.service.auth.IotDeviceTokenService;
  13 +import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
  14 +import lombok.RequiredArgsConstructor;
  15 +import lombok.extern.slf4j.Slf4j;
  16 +import org.eclipse.californium.core.coap.CoAP;
  17 +import org.eclipse.californium.core.coap.MediaTypeRegistry;
  18 +import org.eclipse.californium.core.server.resources.CoapExchange;
  19 +import org.springframework.stereotype.Component;
  20 +
  21 +import java.util.List;
  22 +
  23 +import static cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants.*;
  24 +
  25 +/**
  26 + * IoT 网关 CoAP 协议的【上行】处理器
  27 + *
  28 + * 处理设备通过 CoAP 协议发送的上行消息，包括：
  29 + * 1. 属性上报：POST /topic/sys/{productKey}/{deviceName}/thing/property/post
  30 + * 2. 事件上报：POST /topic/sys/{productKey}/{deviceName}/thing/event/{eventId}/post
  31 + *
  32 + * Token 通过自定义 CoAP Option 2088 携带
  33 + *
  34 + * @author 芋道源码
  35 + */
  36 +@Component
  37 +@RequiredArgsConstructor
  38 +@Slf4j
  39 +public class IotCoapUpstreamHandler {
  40 +
  41 + /**
  42 + * 自定义 CoAP Option 编号，用于携带 Token
  43 + * CoAP Option 范围 2048-65535 属于实验/自定义范围
  44 + */
  45 + public static final int OPTION_TOKEN = 2088;
  46 +
  47 + private final IotDeviceTokenService deviceTokenService;
  48 + private final IotDeviceMessageService deviceMessageService;
  49 +
  50 + /**
  51 + * 处理 CoAP 请求
  52 + *
  53 + * @param exchange CoAP 交换对象
  54 + * @param httpMethod HTTP 方法
  55 + * @param protocol 协议对象
  56 + */
  57 + public void handle(CoapExchange exchange, String httpMethod, IotCoapUpstreamProtocol protocol) {
  58 + try {
  59 + // TODO @AI：这种路径的解析，不用了，简化下，类似 IotHttpUpstreamHandler 这种就很简洁；
  60 + // 1. 解析 URI 路径：/topic/sys/{productKey}/{deviceName}/thing/...
  61 + // 完整路径是 [topic, sys, productKey, deviceName, thing, ...]
  62 + List<String> uriPath = exchange.getRequestOptions().getUriPath();
  63 + if (uriPath.size() < 6) {
  64 + respondError(exchange, CoAP.ResponseCode.BAD_REQUEST,
  65 + "URI 路径格式错误，期望：/topic/sys/{productKey}/{deviceName}/...");
  66 + return;
  67 + }
  68 +
  69 + // 验证路径格式：第一个应该是 "topic"，第二个应该是 "sys"
  70 + if (!"topic".equals(uriPath.get(0)) || !"sys".equals(uriPath.get(1))) {
  71 + respondError(exchange, CoAP.ResponseCode.BAD_REQUEST, "URI 路径格式错误，期望以 /topic/sys 开头");
  72 + return;
  73 + }
  74 +
  75 + // 解析 productKey 和 deviceName（索引 2 和 3）
  76 + String productKey = uriPath.get(2);
  77 + String deviceName = uriPath.get(3);
  78 +
  79 + // 2. 认证：优先从自定义 Option 获取 token，兼容 Query 参数
  80 + String token = getTokenFromOption(exchange);
  81 + if (StrUtil.isEmpty(token)) {
  82 + // 兼容 Query 参数方式
  83 + // TODO @AI：不用兼容 query，简化下；
  84 + token = getQueryParam(exchange, "token");
  85 + }
  86 + if (StrUtil.isEmpty(token)) {
  87 + respondError(exchange, CoAP.ResponseCode.UNAUTHORIZED, "缺少 token（请使用 Option " + OPTION_TOKEN + " 或 Query 参数携带）");
  88 + return;
  89 + }
  90 +
  91 + // 验证 token
  92 + // TODO @AI：这里参考 IotHttpAbstractHandler 简化点校验；
  93 + IotDeviceAuthUtils.DeviceInfo deviceInfo = deviceTokenService.verifyToken(token);
  94 + if (deviceInfo == null) {
  95 + respondError(exchange, CoAP.ResponseCode.UNAUTHORIZED, "token 无效或已过期");
  96 + return;
  97 + }
  98 + // 验证设备信息匹配
  99 + if (ObjUtil.notEqual(productKey, deviceInfo.getProductKey())
  100 + || ObjUtil.notEqual(deviceName, deviceInfo.getDeviceName())) {
  101 + respondError(exchange, CoAP.ResponseCode.FORBIDDEN, "设备信息与 token 不匹配");
  102 + return;
  103 + }
  104 +
  105 + // 3. 解析 method：将 URI 路径转换为 method 格式
  106 + // /topic/sys/pk/dn/thing/property/post -> thing.property.post
  107 + // 路径是 [sys, pk, dn, thing, property, post]，从索引 3 开始
  108 + String method = buildMethod(uriPath);
  109 +
  110 + // 4. 解析并处理消息体
  111 + byte[] payload = exchange.getRequestPayload();
  112 + if (payload == null || payload.length == 0) {
  113 + respondError(exchange, CoAP.ResponseCode.BAD_REQUEST, "请求体不能为空");
  114 + return;
  115 + }
  116 +
  117 + // 5. 解码消息
  118 + IotDeviceMessage message;
  119 + try {
  120 + message = deviceMessageService.decodeDeviceMessage(payload, productKey, deviceName);
  121 + } catch (Exception e) {
  122 + log.error("[handle][消息解码失败]", e);
  123 + respondError(exchange, CoAP.ResponseCode.BAD_REQUEST, "消息解码失败：" + e.getMessage());
  124 + return;
  125 + }
  126 +
  127 + // 校验 method
  128 + // TODO @AI：不用校验 method；以 message 解析出来的为主；
  129 + if (!method.equals(message.getMethod())) {
  130 + log.warn("[handle][method 不匹配，URI: {}, 消息: {}]", method, message.getMethod());
  131 + }
  132 +
  133 + // 6. 发送消息到消息总线
  134 + deviceMessageService.sendDeviceMessage(message, productKey, deviceName, protocol.getServerId());
  135 +
  136 + // 7. 返回成功响应
  137 + respondSuccess(exchange, message.getId());
  138 + } catch (Exception e) {
  139 + log.error("[handle][CoAP 请求处理异常]", e);
  140 + respondError(exchange, CoAP.ResponseCode.INTERNAL_SERVER_ERROR, "服务器内部错误");
  141 + }
  142 + }
  143 +
  144 + /**
  145 + * 构建 method 字符串
  146 + *
  147 + * 将 URI 路径转换为 method 格式，例如：
  148 + * [sys, pk, dn, thing, property, post] -> thing.property.post
  149 + *
  150 + * @param uriPath URI 路径列表
  151 + * @return method 字符串
  152 + */
  153 + private String buildMethod(List<String> uriPath) {
  154 + // 跳过 sys, productKey, deviceName，从第4个元素开始
  155 + if (uriPath.size() > 3) {
  156 + return String.join(StrPool.DOT, uriPath.subList(3, uriPath.size()));
  157 + }
  158 + return "";
  159 + }
  160 +
  161 + // TODO @AI：抽到 coap 的 util 里；
  162 + /**
  163 + * 从自定义 CoAP Option 中获取 Token
  164 + *
  165 + * @param exchange CoAP 交换对象
  166 + * @return Token 值，如果不存在则返回 null
  167 + */
  168 + private String getTokenFromOption(CoapExchange exchange) {
  169 + // 尝试从自定义 Option 2088 获取 Token
  170 + byte[] tokenBytes = exchange.getRequestOptions().getOthers().stream()
  171 + .filter(option -> option.getNumber() == OPTION_TOKEN)
  172 + .findFirst()
  173 + .map(option -> option.getValue())
  174 + .orElse(null);
  175 + if (tokenBytes != null) {
  176 + return new String(tokenBytes);
  177 + }
  178 + return null;
  179 + }
  180 +
  181 + // TODO @AI：抽到 coap 的 util 里；
  182 + /**
  183 + * 从 URI Query 参数中获取指定 key 的值
  184 + *
  185 + * @param exchange CoAP 交换对象
  186 + * @param key 参数名
  187 + * @return 参数值，如果不存在则返回 null
  188 + */
  189 + private String getQueryParam(CoapExchange exchange, String key) {
  190 + for (String query : exchange.getRequestOptions().getUriQuery()) {
  191 + if (query.startsWith(key + "=")) {
  192 + return query.substring((key + "=").length());
  193 + }
  194 + }
  195 + return null;
  196 + }
  197 +
  198 + // TODO @AI：抽到 coap 的 util 里；
  199 + /**
  200 + * 返回成功响应
  201 + *
  202 + * @param exchange CoAP 交换对象
  203 + * @param messageId 消息 ID
  204 + */
  205 + private void respondSuccess(CoapExchange exchange, String messageId) {
  206 + CommonResult<Object> result = CommonResult.success(MapUtil.of("messageId", messageId));
  207 + String json = JsonUtils.toJsonString(result);
  208 + exchange.respond(CoAP.ResponseCode.CONTENT, json, MediaTypeRegistry.APPLICATION_JSON);
  209 + }
  210 +
  211 + // TODO @AI：抽到 coap 的 util 里；
  212 + /**
  213 + * 返回错误响应
  214 + *
  215 + * @param exchange CoAP 交换对象
  216 + * @param code CoAP 响应码
  217 + * @param message 错误消息
  218 + */
  219 + private void respondError(CoapExchange exchange, CoAP.ResponseCode code, String message) {
  220 + // 将 CoAP 响应码映射到业务错误码
  221 + int errorCode = mapCoapCodeToErrorCode(code);
  222 + CommonResult<Object> result = CommonResult.error(errorCode, message);
  223 + String json = JsonUtils.toJsonString(result);
  224 + exchange.respond(code, json, MediaTypeRegistry.APPLICATION_JSON);
  225 + }
  226 +
  227 + // TODO @AI：兼容 jdk8 的写法；
  228 + /**
  229 + * 将 CoAP 响应码映射到业务错误码
  230 + *
  231 + * @param code CoAP 响应码
  232 + * @return 业务错误码
  233 + */
  234 + private int mapCoapCodeToErrorCode(CoAP.ResponseCode code) {
  235 + return switch (code) {
  236 + case BAD_REQUEST -> BAD_REQUEST.getCode();
  237 + case UNAUTHORIZED -> UNAUTHORIZED.getCode();
  238 + case FORBIDDEN -> FORBIDDEN.getCode();
  239 + default -> INTERNAL_SERVER_ERROR.getCode();
  240 + };
  241 + }
  242 +
  243 +}
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/router/IotCoapUpstreamTopicResource.java 0 → 100644
  View file @bec8cc6
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.coap.router;
  2 +
  3 +import cn.iocoder.yudao.module.iot.gateway.protocol.coap.IotCoapUpstreamProtocol;
  4 +import lombok.extern.slf4j.Slf4j;
  5 +import org.eclipse.californium.core.CoapResource;
  6 +import org.eclipse.californium.core.server.resources.CoapExchange;
  7 +import org.eclipse.californium.core.server.resources.Resource;
  8 +
  9 +/**
  10 + * IoT 网关 CoAP 协议的【上行】Topic 资源
  11 + *
  12 + * 支持任意深度的路径匹配：
  13 + * - /topic/sys/{productKey}/{deviceName}/thing/property/post
  14 + * - /topic/sys/{productKey}/{deviceName}/thing/event/{eventId}/post
  15 + *
  16 + * @author 芋道源码
  17 + */
  18 +@Slf4j
  19 +public class IotCoapUpstreamTopicResource extends CoapResource {
  20 +
  21 + public static final String PATH = "topic";
  22 +
  23 + private final IotCoapUpstreamProtocol protocol;
  24 + private final IotCoapUpstreamHandler upstreamHandler;
  25 +
  26 + /**
  27 + * 创建根资源（/topic）
  28 + */
  29 + public IotCoapUpstreamTopicResource(IotCoapUpstreamProtocol protocol,
  30 + IotCoapUpstreamHandler upstreamHandler) {
  31 + this(PATH, protocol, upstreamHandler);
  32 + log.info("[IotCoapUpstreamTopicResource][创建 CoAP 上行 Topic 资源: /{}]", PATH);
  33 + }
  34 +
  35 + /**
  36 + * 创建子资源（动态路径）
  37 + */
  38 + private IotCoapUpstreamTopicResource(String name,
  39 + IotCoapUpstreamProtocol protocol,
  40 + IotCoapUpstreamHandler upstreamHandler) {
  41 + super(name);
  42 + this.protocol = protocol;
  43 + this.upstreamHandler = upstreamHandler;
  44 + }
  45 +
  46 + @Override
  47 + public Resource getChild(String name) {
  48 + // 递归创建动态子资源，支持任意深度路径
  49 + return new IotCoapUpstreamTopicResource(name, protocol, upstreamHandler);
  50 + }
  51 +
  52 + @Override
  53 + public void handleGET(CoapExchange exchange) {
  54 + upstreamHandler.handle(exchange, "GET", protocol);
  55 + }
  56 +
  57 + @Override
  58 + public void handlePOST(CoapExchange exchange) {
  59 + upstreamHandler.handle(exchange, "POST", protocol);
  60 + }
  61 +
  62 + @Override
  63 + public void handlePUT(CoapExchange exchange) {
  64 + upstreamHandler.handle(exchange, "PUT", protocol);
  65 + }
  66 +
  67 +}
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/package-info.java 0 → 100644
  View file @bec8cc6
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.emqx;
0 \ No newline at end of file 2 \ No newline at end of file
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/http/package-info.java 0 → 100644
  View file @bec8cc6
  1 +// TODO @AI：参考 /Users/yunai/Java/ruoyi-vue-pro-jdk25/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/package-info.java 完善注释；
  2 +package cn.iocoder.yudao.module.iot.gateway.protocol.http;
0 \ No newline at end of file 3 \ No newline at end of file
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/package-info.java 0 → 100644
  View file @bec8cc6
  1 +// TODO @AI：参考 /Users/yunai/Java/ruoyi-vue-pro-jdk25/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/package-info.java
  2 +package cn.iocoder.yudao.module.iot.gateway.protocol.tcp;
0 \ No newline at end of file 3 \ No newline at end of file
yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application.yaml
  View file @bec8cc6
@@ -117,6 +117,15 @@ yudao: @@ -117,6 +117,15 @@ yudao:
117 keep-alive-timeout-seconds: 300 # 保持连接超时时间（秒） 117 keep-alive-timeout-seconds: 300 # 保持连接超时时间（秒）
118 ssl-enabled: false # 是否启用 SSL（wss://） 118 ssl-enabled: false # 是否启用 SSL（wss://）
119 sub-protocol: mqtt # WebSocket 子协议 119 sub-protocol: mqtt # WebSocket 子协议
  120 + # ====================================
  121 + # 针对引入的 CoAP 组件的配置
  122 + # ====================================
  123 + coap:
  124 + enabled: false # 是否启用 CoAP 协议
  125 + port: 5683 # CoAP 服务端口（默认 5683）
  126 + max-message-size: 1024 # 最大消息大小（字节）
  127 + ack-timeout: 2000 # ACK 超时时间（毫秒）
  128 + max-retransmit: 4 # 最大重传次数
120 129
121 --- #################### 日志相关配置 #################### 130 --- #################### 日志相关配置 ####################
122 131
@@ -137,6 +146,7 @@ logging: @@ -137,6 +146,7 @@ logging:
137 cn.iocoder.yudao.module.iot.gateway.protocol.http: DEBUG 146 cn.iocoder.yudao.module.iot.gateway.protocol.http: DEBUG
138 cn.iocoder.yudao.module.iot.gateway.protocol.mqtt: DEBUG 147 cn.iocoder.yudao.module.iot.gateway.protocol.mqtt: DEBUG
139 cn.iocoder.yudao.module.iot.gateway.protocol.mqttws: DEBUG 148 cn.iocoder.yudao.module.iot.gateway.protocol.mqttws: DEBUG
  149 + cn.iocoder.yudao.module.iot.gateway.protocol.coap: DEBUG
140 # 根日志级别 150 # 根日志级别
141 root: INFO 151 root: INFO
142 152
yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/IotCoapProtocolIntegrationTest.java 0 → 100644
  View file @bec8cc6
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.coap;
  2 +
  3 +import cn.iocoder.yudao.module.iot.gateway.protocol.coap.router.IotCoapUpstreamHandler;
  4 +import lombok.extern.slf4j.Slf4j;
  5 +import org.eclipse.californium.core.CoapClient;
  6 +import org.eclipse.californium.core.CoapResponse;
  7 +import org.eclipse.californium.core.coap.MediaTypeRegistry;
  8 +import org.eclipse.californium.core.coap.Option;
  9 +import org.eclipse.californium.core.coap.Request;
  10 +import org.junit.jupiter.api.*;
  11 +
  12 +/**
  13 + * IoT 网关 CoAP 协议集成测试（手动测试）
  14 + *
  15 + * 使用步骤：
  16 + * 1. 启动 CoAP 网关服务（端口 5683）
  17 + * 2. 运行 testAuth() 获取 token
  18 + * 3. 将 token 粘贴到 TOKEN 常量
  19 + * 4. 运行 testPropertyPost() 或 testEventPost()
  20 + *
  21 + * @author 芋道源码
  22 + */
  23 +@Slf4j
  24 +@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
  25 +class IotCoapProtocolIntegrationTest {
  26 +
  27 + private static final String SERVER_HOST = "127.0.0.1";
  28 + private static final int SERVER_PORT = 5683;
  29 +
  30 + // 设备信息（根据实际情况修改）
  31 + private static final String PRODUCT_KEY = "testProductKey";
  32 + private static final String DEVICE_NAME = "testDeviceName";
  33 + private static final String CLIENT_ID = PRODUCT_KEY + "." + DEVICE_NAME;
  34 + private static final String USERNAME = DEVICE_NAME + "&" + PRODUCT_KEY;
  35 + private static final String PASSWORD = "testPassword123";
  36 +
  37 + // TODO: 运行 testAuth() 后，将返回的 token 粘贴到这里
  38 + private static final String TOKEN = "粘贴你的token到这里";
  39 +
  40 + // ========== 1. 认证测试 ==========
  41 +
  42 + @Test
  43 + @Order(1)
  44 + @DisplayName("1. 认证 - 获取 Token")
  45 + void testAuth() throws Exception {
  46 + String uri = String.format("coap://%s:%d/auth", SERVER_HOST, SERVER_PORT);
  47 +
  48 + String payload = String.format("""
  49 + {
  50 + "clientId": "%s",
  51 + "username": "%s",
  52 + "password": "%s"
  53 + }
  54 + """, CLIENT_ID, USERNAME, PASSWORD);
  55 +
  56 + CoapClient client = new CoapClient(uri);
  57 + try {
  58 + log.info("[testAuth][请求 URI: {}]", uri);
  59 + log.info("[testAuth][请求体: {}]", payload);
  60 +
  61 + CoapResponse response = client.post(payload, MediaTypeRegistry.APPLICATION_JSON);
  62 +
  63 + log.info("[testAuth][响应码: {}]", response.getCode());
  64 + log.info("[testAuth][响应体: {}]", response.getResponseText());
  65 + log.info("[testAuth][请将返回的 token 复制到 TOKEN 常量中]");
  66 + } finally {
  67 + client.shutdown();
  68 + }
  69 + }
  70 +
  71 + // ========== 2. 属性上报测试 ==========
  72 +
  73 + @Test
  74 + @Order(2)
  75 + @DisplayName("2. 属性上报")
  76 + void testPropertyPost() throws Exception {
  77 + String uri = String.format("coap://%s:%d/topic/sys/%s/%s/thing/property/post",
  78 + SERVER_HOST, SERVER_PORT, PRODUCT_KEY, DEVICE_NAME);
  79 +
  80 + String payload = """
  81 + {
  82 + "id": "123",
  83 + "method": "thing.property.post",
  84 + "params": {
  85 + "temperature": 25.5,
  86 + "humidity": 60
  87 + }
  88 + }
  89 + """;
  90 +
  91 + CoapClient client = new CoapClient(uri);
  92 + try {
  93 + // 构造带自定义 Option 的请求
  94 + Request request = Request.newPost();
  95 + request.setURI(uri);
  96 + request.setPayload(payload);
  97 + request.getOptions().setContentFormat(MediaTypeRegistry.APPLICATION_JSON);
  98 + // 添加自定义 Token Option (2088)
  99 + request.getOptions().addOption(new Option(IotCoapUpstreamHandler.OPTION_TOKEN, TOKEN));
  100 +
  101 + log.info("[testPropertyPost][请求 URI: {}]", uri);
  102 + log.info("[testPropertyPost][Token: {}]", TOKEN);
  103 + log.info("[testPropertyPost][请求体: {}]", payload);
  104 +
  105 + CoapResponse response = client.advanced(request);
  106 +
  107 + log.info("[testPropertyPost][响应码: {}]", response.getCode());
  108 + log.info("[testPropertyPost][响应体: {}]", response.getResponseText());
  109 + } finally {
  110 + client.shutdown();
  111 + }
  112 + }
  113 +
  114 + // ========== 3. 事件上报测试 ==========
  115 +
  116 + @Test
  117 + @Order(3)
  118 + @DisplayName("3. 事件上报")
  119 + void testEventPost() throws Exception {
  120 + String uri = String.format("coap://%s:%d/topic/sys/%s/%s/thing/event/alarm/post",
  121 + SERVER_HOST, SERVER_PORT, PRODUCT_KEY, DEVICE_NAME);
  122 +
  123 + String payload = """
  124 + {
  125 + "id": "456",
  126 + "method": "thing.event.alarm.post",
  127 + "params": {
  128 + "alarmType": "temperature_high",
  129 + "level": "warning",
  130 + "value": 85.2
  131 + }
  132 + }
  133 + """;
  134 +
  135 + CoapClient client = new CoapClient(uri);
  136 + try {
  137 + // 构造带自定义 Option 的请求
  138 + Request request = Request.newPost();
  139 + request.setURI(uri);
  140 + request.setPayload(payload);
  141 + request.getOptions().setContentFormat(MediaTypeRegistry.APPLICATION_JSON);
  142 + // 添加自定义 Token Option (2088)
  143 + request.getOptions().addOption(new Option(IotCoapUpstreamHandler.OPTION_TOKEN, TOKEN));
  144 +
  145 + log.info("[testEventPost][请求 URI: {}]", uri);
  146 + log.info("[testEventPost][Token: {}]", TOKEN);
  147 + log.info("[testEventPost][请求体: {}]", payload);
  148 +
  149 + CoapResponse response = client.advanced(request);
  150 +
  151 + log.info("[testEventPost][响应码: {}]", response.getCode());
  152 + log.info("[testEventPost][响应体: {}]", response.getResponseText());
  153 + } finally {
  154 + client.shutdown();
  155 + }
  156 + }
  157 +
  158 +}