BJ_ZTEITS-Developers / parking-cloud-yudao

Commit eb5fa9fd914b3620dd7948092fa5c98b1c404839

Authored by 芋道源码
Committed by Gitee
2 parents 8a2c38a0 52b8e664

!1500 feat：【iot】udp 协议 100%：完善注释、完善单测 

Merge pull request !1500 from 芋道源码/feature/iot-udp
Inline Side-by-side
Showing 12 changed files with 1180 additions and 2 deletions
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayConfiguration.java
  View file @eb5fa9f
... ... @@ -17,6 +17,9 @@ import cn.iocoder.yudao.module.iot.gateway.protocol.mqttws.router.IotMqttWsDowns
17 17 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.IotTcpDownstreamSubscriber;
18 18 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.IotTcpUpstreamProtocol;
19 19 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.manager.IotTcpConnectionManager;
  20 +import cn.iocoder.yudao.module.iot.gateway.protocol.udp.IotUdpDownstreamSubscriber;
  21 +import cn.iocoder.yudao.module.iot.gateway.protocol.udp.IotUdpUpstreamProtocol;
  22 +import cn.iocoder.yudao.module.iot.gateway.protocol.udp.manager.IotUdpSessionManager;
20 23 import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceService;
21 24 import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
22 25 import io.vertx.core.Vertx;
... ... @@ -194,4 +197,39 @@ public class IotGatewayConfiguration {
194 197  
195 198 }
196 199  
  200 + /**
  201 + * IoT 网关 UDP 协议配置类
  202 + */
  203 + @Configuration
  204 + @ConditionalOnProperty(prefix = "yudao.iot.gateway.protocol.udp", name = "enabled", havingValue = "true")
  205 + @Slf4j
  206 + public static class UdpProtocolConfiguration {
  207 +
  208 + @Bean(name = "udpVertx", destroyMethod = "close")
  209 + public Vertx udpVertx() {
  210 + return Vertx.vertx();
  211 + }
  212 +
  213 + @Bean
  214 + public IotUdpUpstreamProtocol iotUdpUpstreamProtocol(IotGatewayProperties gatewayProperties,
  215 + IotDeviceService deviceService,
  216 + IotDeviceMessageService messageService,
  217 + IotUdpSessionManager sessionManager,
  218 + @Qualifier("udpVertx") Vertx udpVertx) {
  219 + return new IotUdpUpstreamProtocol(gatewayProperties.getProtocol().getUdp(),
  220 + deviceService, messageService, sessionManager, udpVertx);
  221 + }
  222 +
  223 + @Bean
  224 + public IotUdpDownstreamSubscriber iotUdpDownstreamSubscriber(IotUdpUpstreamProtocol protocolHandler,
  225 + IotDeviceMessageService messageService,
  226 + IotDeviceService deviceService,
  227 + IotUdpSessionManager sessionManager,
  228 + IotMessageBus messageBus) {
  229 + return new IotUdpDownstreamSubscriber(protocolHandler, messageService, deviceService, sessionManager,
  230 + messageBus);
  231 + }
  232 +
  233 + }
  234 +
197 235 }
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayProperties.java
  View file @eb5fa9f
... ... @@ -93,6 +93,11 @@ public class IotGatewayProperties {
93 93 */
94 94 private MqttWsProperties mqttWs;
95 95  
  96 + /**
  97 + * UDP 组件配置
  98 + */
  99 + private UdpProperties udp;
  100 +
96 101 }
97 102  
98 103 @Data
... ... @@ -503,4 +508,42 @@ public class IotGatewayProperties {
503 508  
504 509 }
505 510  
  511 + @Data
  512 + public static class UdpProperties {
  513 +
  514 + /**
  515 + * 是否开启
  516 + */
  517 + @NotNull(message = "是否开启不能为空")
  518 + private Boolean enabled;
  519 +
  520 + /**
  521 + * 服务端口（默认 8092）
  522 + */
  523 + private Integer port = 8092;
  524 +
  525 + /**
  526 + * 接收缓冲区大小（默认 64KB）
  527 + */
  528 + private Integer receiveBufferSize = 65536;
  529 +
  530 + /**
  531 + * 发送缓冲区大小（默认 64KB）
  532 + */
  533 + private Integer sendBufferSize = 65536;
  534 +
  535 + /**
  536 + * 会话超时时间（毫秒，默认 60 秒）
  537 + * <p>
  538 + * 用于清理不活跃的设备地址映射
  539 + */
  540 + private Long sessionTimeoutMs = 60000L;
  541 +
  542 + /**
  543 + * 会话清理间隔（毫秒，默认 30 秒）
  544 + */
  545 + private Long sessionCleanIntervalMs = 30000L;
  546 +
  547 + }
  548 +
506 549 }
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/manager/IotTcpConnectionManager.java
  View file @eb5fa9f
1 1 package cn.iocoder.yudao.module.iot.gateway.protocol.tcp.manager;
2 2  
  3 +import io.vertx.core.buffer.Buffer;
3 4 import io.vertx.core.net.NetSocket;
4 5 import lombok.Data;
5 6 import lombok.extern.slf4j.Slf4j;
... ... @@ -119,7 +120,7 @@ public class IotTcpConnectionManager {
119 120 }
120 121  
121 122 try {
122   - socket.write(io.vertx.core.buffer.Buffer.buffer(data));
  123 + socket.write(Buffer.buffer(data));
123 124 log.debug("[sendToDevice][发送消息成功，设备 ID: {}，数据长度: {} 字节]", deviceId, data.length);
124 125 return true;
125 126 } catch (Exception e) {
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/router/IotTcpDownstreamHandler.java
  View file @eb5fa9f
... ... @@ -43,7 +43,7 @@ public class IotTcpDownstreamHandler {
43 43 return;
44 44 }
45 45  
46   - // 2. 根据产品 Key 和设备名称编码消息并发送到设备
  46 + // 2. 根据产品 Key 和设备名称编码消息，并发送到设备
47 47 byte[] bytes = deviceMessageService.encodeDeviceMessage(message, deviceInfo.getProductKey(),
48 48 deviceInfo.getDeviceName());
49 49 boolean success = connectionManager.sendToDevice(message.getDeviceId(), bytes);
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/udp/IotUdpDownstreamSubscriber.java 0 → 100644
  View file @eb5fa9f
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.udp;
  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 cn.iocoder.yudao.module.iot.gateway.protocol.udp.manager.IotUdpSessionManager;
  8 +import cn.iocoder.yudao.module.iot.gateway.protocol.udp.router.IotUdpDownstreamHandler;
  9 +import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceService;
  10 +import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
  11 +import jakarta.annotation.PostConstruct;
  12 +import lombok.RequiredArgsConstructor;
  13 +import lombok.extern.slf4j.Slf4j;
  14 +
  15 +/**
  16 + * IoT 网关 UDP 下游订阅者：接收下行给设备的消息
  17 + *
  18 + * @author 芋道源码
  19 + */
  20 +@Slf4j
  21 +@RequiredArgsConstructor
  22 +public class IotUdpDownstreamSubscriber implements IotMessageSubscriber<IotDeviceMessage> {
  23 +
  24 + private final IotUdpUpstreamProtocol protocol;
  25 +
  26 + private final IotDeviceMessageService messageService;
  27 +
  28 + private final IotDeviceService deviceService;
  29 +
  30 + private final IotUdpSessionManager sessionManager;
  31 +
  32 + private final IotMessageBus messageBus;
  33 +
  34 + private IotUdpDownstreamHandler downstreamHandler;
  35 +
  36 + @PostConstruct
  37 + public void init() {
  38 + // 初始化下游处理器
  39 + this.downstreamHandler = new IotUdpDownstreamHandler(messageService, deviceService, sessionManager, protocol);
  40 +
  41 + messageBus.register(this);
  42 + log.info("[init][UDP 下游订阅者初始化完成，服务器 ID: {}，Topic: {}]",
  43 + protocol.getServerId(), getTopic());
  44 + }
  45 +
  46 + @Override
  47 + public String getTopic() {
  48 + return IotDeviceMessageUtils.buildMessageBusGatewayDeviceMessageTopic(protocol.getServerId());
  49 + }
  50 +
  51 + @Override
  52 + public String getGroup() {
  53 + // 保证点对点消费，需要保证独立的 Group，所以使用 Topic 作为 Group
  54 + return getTopic();
  55 + }
  56 +
  57 + @Override
  58 + public void onMessage(IotDeviceMessage message) {
  59 + try {
  60 + downstreamHandler.handle(message);
  61 + } catch (Exception e) {
  62 + log.error("[onMessage][处理下行消息失败，设备 ID: {}，方法: {}，消息 ID: {}]",
  63 + message.getDeviceId(), message.getMethod(), message.getId(), e);
  64 + }
  65 + }
  66 +
  67 +}
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/udp/IotUdpUpstreamProtocol.java 0 → 100644
  View file @eb5fa9f
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.udp;
  2 +
  3 +import cn.hutool.core.collection.CollUtil;
  4 +import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceRespDTO;
  5 +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
  6 +import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
  7 +import cn.iocoder.yudao.module.iot.gateway.config.IotGatewayProperties;
  8 +import cn.iocoder.yudao.module.iot.gateway.protocol.udp.manager.IotUdpSessionManager;
  9 +import cn.iocoder.yudao.module.iot.gateway.protocol.udp.router.IotUdpUpstreamHandler;
  10 +import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceService;
  11 +import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
  12 +import io.vertx.core.Vertx;
  13 +import io.vertx.core.datagram.DatagramSocket;
  14 +import io.vertx.core.datagram.DatagramSocketOptions;
  15 +import jakarta.annotation.PostConstruct;
  16 +import jakarta.annotation.PreDestroy;
  17 +import lombok.Getter;
  18 +import lombok.extern.slf4j.Slf4j;
  19 +
  20 +import java.util.List;
  21 +
  22 +/**
  23 + * IoT 网关 UDP 协议：接收设备上行消息
  24 + * <p>
  25 + * 采用 Vertx DatagramSocket 实现 UDP 服务器，主要功能：
  26 + * 1. 监听 UDP 端口，接收设备消息
  27 + * 2. 定期清理不活跃的设备地址映射
  28 + * 3. 提供 UDP Socket 用于下行消息发送
  29 + *
  30 + * @author 芋道源码
  31 + */
  32 +@Slf4j
  33 +public class IotUdpUpstreamProtocol {
  34 +
  35 + private final IotGatewayProperties.UdpProperties udpProperties;
  36 +
  37 + private final IotDeviceService deviceService;
  38 +
  39 + private final IotDeviceMessageService messageService;
  40 +
  41 + private final IotUdpSessionManager sessionManager;
  42 +
  43 + private final Vertx vertx;
  44 +
  45 + @Getter
  46 + private final String serverId;
  47 +
  48 + @Getter
  49 + private DatagramSocket udpSocket;
  50 +
  51 + /**
  52 + * 会话清理定时器 ID
  53 + */
  54 + private Long cleanTimerId;
  55 +
  56 + private IotUdpUpstreamHandler upstreamHandler;
  57 +
  58 + public IotUdpUpstreamProtocol(IotGatewayProperties.UdpProperties udpProperties,
  59 + IotDeviceService deviceService,
  60 + IotDeviceMessageService messageService,
  61 + IotUdpSessionManager sessionManager,
  62 + Vertx vertx) {
  63 + this.udpProperties = udpProperties;
  64 + this.deviceService = deviceService;
  65 + this.messageService = messageService;
  66 + this.sessionManager = sessionManager;
  67 + this.vertx = vertx;
  68 + this.serverId = IotDeviceMessageUtils.generateServerId(udpProperties.getPort());
  69 + }
  70 +
  71 + @PostConstruct
  72 + public void start() {
  73 + // 1. 初始化上行消息处理器
  74 + this.upstreamHandler = new IotUdpUpstreamHandler(this, messageService, deviceService, sessionManager);
  75 +
  76 + // 2. 创建 UDP Socket 选项
  77 + DatagramSocketOptions options = new DatagramSocketOptions()
  78 + .setReceiveBufferSize(udpProperties.getReceiveBufferSize())
  79 + .setSendBufferSize(udpProperties.getSendBufferSize())
  80 + .setReuseAddress(true);
  81 +
  82 + // 3. 创建 UDP Socket
  83 + udpSocket = vertx.createDatagramSocket(options);
  84 +
  85 + // 4. 监听端口
  86 + udpSocket.listen(udpProperties.getPort(), "0.0.0.0", result -> {
  87 + if (result.failed()) {
  88 + log.error("[start][IoT 网关 UDP 协议启动失败]", result.cause());
  89 + return;
  90 + }
  91 + // 设置数据包处理器
  92 + udpSocket.handler(packet -> upstreamHandler.handle(packet, udpSocket));
  93 + log.info("[start][IoT 网关 UDP 协议启动成功，端口：{}，接收缓冲区：{} 字节，发送缓冲区：{} 字节]",
  94 + udpProperties.getPort(), udpProperties.getReceiveBufferSize(),
  95 + udpProperties.getSendBufferSize());
  96 +
  97 + // 5. 启动会话清理定时器
  98 + startSessionCleanTimer();
  99 + });
  100 + }
  101 +
  102 + @PreDestroy
  103 + public void stop() {
  104 + // 1. 取消会话清理定时器
  105 + if (cleanTimerId != null) {
  106 + vertx.cancelTimer(cleanTimerId);
  107 + cleanTimerId = null;
  108 + log.info("[stop][会话清理定时器已取消]");
  109 + }
  110 +
  111 + // 2. 关闭 UDP Socket
  112 + if (udpSocket != null) {
  113 + try {
  114 + udpSocket.close().result();
  115 + log.info("[stop][IoT 网关 UDP 协议已停止]");
  116 + } catch (Exception e) {
  117 + log.error("[stop][IoT 网关 UDP 协议停止失败]", e);
  118 + }
  119 + }
  120 + }
  121 +
  122 + /**
  123 + * 启动会话清理定时器
  124 + */
  125 + private void startSessionCleanTimer() {
  126 + cleanTimerId = vertx.setPeriodic(udpProperties.getSessionCleanIntervalMs(), id -> {
  127 + try {
  128 + // 1. 清理超时的设备地址映射，并获取离线设备列表
  129 + List<Long> offlineDeviceIds = sessionManager.cleanExpiredMappings(udpProperties.getSessionTimeoutMs());
  130 +
  131 + // 2. 为每个离线设备发送离线消息
  132 + for (Long deviceId : offlineDeviceIds) {
  133 + sendOfflineMessage(deviceId);
  134 + }
  135 + if (CollUtil.isNotEmpty(offlineDeviceIds)) {
  136 + log.info("[cleanExpiredMappings][本次清理 {} 个超时设备]", offlineDeviceIds.size());
  137 + }
  138 + } catch (Exception e) {
  139 + log.error("[cleanExpiredMappings][清理超时会话失败]", e);
  140 + }
  141 + });
  142 + log.info("[startSessionCleanTimer][会话清理定时器启动，间隔：{} ms，超时：{} ms]",
  143 + udpProperties.getSessionCleanIntervalMs(), udpProperties.getSessionTimeoutMs());
  144 + }
  145 +
  146 + /**
  147 + * 发送设备离线消息
  148 + *
  149 + * @param deviceId 设备 ID
  150 + */
  151 + private void sendOfflineMessage(Long deviceId) {
  152 + try {
  153 + // 获取设备信息
  154 + IotDeviceRespDTO device = deviceService.getDeviceFromCache(deviceId);
  155 + if (device == null) {
  156 + log.warn("[sendOfflineMessage][设备不存在，设备 ID: {}]", deviceId);
  157 + return;
  158 + }
  159 +
  160 + // 发送离线消息
  161 + IotDeviceMessage offlineMessage = IotDeviceMessage.buildStateOffline();
  162 + messageService.sendDeviceMessage(offlineMessage, device.getProductKey(),
  163 + device.getDeviceName(), serverId);
  164 + log.info("[sendOfflineMessage][发送离线消息，设备 ID: {}，设备名: {}]",
  165 + deviceId, device.getDeviceName());
  166 + } catch (Exception e) {
  167 + log.error("[sendOfflineMessage][发送离线消息失败，设备 ID: {}]", deviceId, e);
  168 + }
  169 + }
  170 +
  171 +}
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/udp/manager/IotUdpSessionManager.java 0 → 100644
  View file @eb5fa9f
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.udp.manager;
  2 +
  3 +import io.vertx.core.buffer.Buffer;
  4 +import io.vertx.core.datagram.DatagramSocket;
  5 +import lombok.extern.slf4j.Slf4j;
  6 +import org.springframework.stereotype.Component;
  7 +
  8 +import java.net.InetSocketAddress;
  9 +import java.time.LocalDateTime;
  10 +import java.util.ArrayList;
  11 +import java.util.Iterator;
  12 +import java.util.List;
  13 +import java.util.Map;
  14 +import java.util.concurrent.ConcurrentHashMap;
  15 +
  16 +/**
  17 + * IoT 网关 UDP 会话管理器
  18 + * <p>
  19 + * 采用无状态设计，SessionManager 主要用于：
  20 + * 1. 管理设备地址映射（用于下行消息发送）
  21 + * 2. 定期清理不活跃的设备地址映射
  22 + * <p>
  23 + * 注意：UDP 是无连接协议，上行消息通过 token 验证身份，不依赖会话状态
  24 + *
  25 + * @author 芋道源码
  26 + */
  27 +@Slf4j
  28 +@Component
  29 +public class IotUdpSessionManager {
  30 +
  31 + /**
  32 + * 设备 ID -> 设备地址（用于下行消息发送）
  33 + */
  34 + private final Map<Long, InetSocketAddress> deviceAddressMap = new ConcurrentHashMap<>();
  35 +
  36 + /**
  37 + * 设备地址 Key -> 最后活跃时间（用于清理）
  38 + */
  39 + private final Map<String, LocalDateTime> lastActiveTimeMap = new ConcurrentHashMap<>();
  40 +
  41 + /**
  42 + * 设备地址 Key -> 设备 ID（反向映射，用于清理时同步）
  43 + */
  44 + private final Map<String, Long> addressDeviceMap = new ConcurrentHashMap<>();
  45 +
  46 + /**
  47 + * 更新设备地址（每次收到上行消息时调用）
  48 + *
  49 + * @param deviceId 设备 ID
  50 + * @param address 设备地址
  51 + */
  52 + public void updateDeviceAddress(Long deviceId, InetSocketAddress address) {
  53 + String addressKey = buildAddressKey(address);
  54 + // 更新设备地址映射
  55 + deviceAddressMap.put(deviceId, address);
  56 + lastActiveTimeMap.put(addressKey, LocalDateTime.now());
  57 + addressDeviceMap.put(addressKey, deviceId);
  58 + log.debug("[updateDeviceAddress][更新设备地址，设备 ID: {}，地址: {}]", deviceId, addressKey);
  59 + }
  60 +
  61 + /**
  62 + * 检查设备是否在线（即是否有地址映射）
  63 + *
  64 + * @param deviceId 设备 ID
  65 + * @return 是否在线
  66 + */
  67 + public boolean isDeviceOnline(Long deviceId) {
  68 + return deviceAddressMap.containsKey(deviceId);
  69 + }
  70 +
  71 + /**
  72 + * 检查设备是否离线
  73 + *
  74 + * @param deviceId 设备 ID
  75 + * @return 是否离线
  76 + */
  77 + public boolean isDeviceOffline(Long deviceId) {
  78 + return !isDeviceOnline(deviceId);
  79 + }
  80 +
  81 + /**
  82 + * 发送消息到设备
  83 + *
  84 + * @param deviceId 设备 ID
  85 + * @param data 数据
  86 + * @param socket UDP Socket
  87 + * @return 是否发送成功
  88 + */
  89 + public boolean sendToDevice(Long deviceId, byte[] data, DatagramSocket socket) {
  90 + InetSocketAddress address = deviceAddressMap.get(deviceId);
  91 + if (address == null) {
  92 + log.warn("[sendToDevice][设备地址不存在，设备 ID: {}]", deviceId);
  93 + return false;
  94 + }
  95 +
  96 + try {
  97 + socket.send(Buffer.buffer(data), address.getPort(), address.getHostString(), result -> {
  98 + if (result.succeeded()) {
  99 + log.debug("[sendToDevice][发送消息成功，设备 ID: {}，地址: {}，数据长度: {} 字节]",
  100 + deviceId, buildAddressKey(address), data.length);
  101 + } else {
  102 + log.error("[sendToDevice][发送消息失败，设备 ID: {}，地址: {}]",
  103 + deviceId, buildAddressKey(address), result.cause());
  104 + }
  105 + });
  106 + return true;
  107 + } catch (Exception e) {
  108 + log.error("[sendToDevice][发送消息异常，设备 ID: {}]", deviceId, e);
  109 + return false;
  110 + }
  111 + }
  112 +
  113 + /**
  114 + * 定期清理不活跃的设备地址映射
  115 + *
  116 + * @param timeoutMs 超时时间（毫秒）
  117 + * @return 清理的设备 ID 列表（用于发送离线消息）
  118 + */
  119 + public List<Long> cleanExpiredMappings(long timeoutMs) {
  120 + List<Long> offlineDeviceIds = new ArrayList<>();
  121 + LocalDateTime now = LocalDateTime.now();
  122 + LocalDateTime expireTime = now.minusNanos(timeoutMs * 1_000_000);
  123 + Iterator<Map.Entry<String, LocalDateTime>> iterator = lastActiveTimeMap.entrySet().iterator();
  124 + while (iterator.hasNext()) {
  125 + // 未过期，跳过
  126 + Map.Entry<String, LocalDateTime> entry = iterator.next();
  127 + if (entry.getValue().isAfter(expireTime)) {
  128 + continue;
  129 + }
  130 + // 过期处理：记录离线设备 ID
  131 + String addressKey = entry.getKey();
  132 + Long deviceId = addressDeviceMap.remove(addressKey);
  133 + if (deviceId == null) {
  134 + iterator.remove();
  135 + continue;
  136 + }
  137 + InetSocketAddress address = deviceAddressMap.remove(deviceId);
  138 + if (address == null) {
  139 + iterator.remove();
  140 + continue;
  141 + }
  142 + offlineDeviceIds.add(deviceId);
  143 + log.debug("[cleanExpiredMappings][清理超时设备，设备 ID: {}，地址: {}，最后活跃时间: {}]",
  144 + deviceId, addressKey, entry.getValue());
  145 + iterator.remove();
  146 + }
  147 + return offlineDeviceIds;
  148 + }
  149 +
  150 + /**
  151 + * 构建地址 Key
  152 + *
  153 + * @param address 地址
  154 + * @return 地址 Key
  155 + */
  156 + public String buildAddressKey(InetSocketAddress address) {
  157 + return address.getHostString() + ":" + address.getPort();
  158 + }
  159 +
  160 +}
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/udp/package-info.java 0 → 100644
  View file @eb5fa9f
  1 +/**
  2 + * UDP 协议实现包
  3 + * <p>
  4 + * 提供基于 Vert.x DatagramSocket 的 IoT 设备连接和消息处理功能
  5 + */
  6 +package cn.iocoder.yudao.module.iot.gateway.protocol.udp;
0 7 \ No newline at end of file
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/udp/router/IotUdpDownstreamHandler.java 0 → 100644
  View file @eb5fa9f
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.udp.router;
  2 +
  3 +import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceRespDTO;
  4 +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
  5 +import cn.iocoder.yudao.module.iot.gateway.protocol.udp.IotUdpUpstreamProtocol;
  6 +import cn.iocoder.yudao.module.iot.gateway.protocol.udp.manager.IotUdpSessionManager;
  7 +import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceService;
  8 +import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
  9 +import io.vertx.core.datagram.DatagramSocket;
  10 +import lombok.extern.slf4j.Slf4j;
  11 +
  12 +/**
  13 + * IoT 网关 UDP 下行消息处理器
  14 + *
  15 + * @author 芋道源码
  16 + */
  17 +@Slf4j
  18 +public class IotUdpDownstreamHandler {
  19 +
  20 + private final IotDeviceMessageService deviceMessageService;
  21 +
  22 + private final IotDeviceService deviceService;
  23 +
  24 + private final IotUdpSessionManager sessionManager;
  25 +
  26 + private final IotUdpUpstreamProtocol protocol;
  27 +
  28 + public IotUdpDownstreamHandler(IotDeviceMessageService deviceMessageService,
  29 + IotDeviceService deviceService,
  30 + IotUdpSessionManager sessionManager,
  31 + IotUdpUpstreamProtocol protocol) {
  32 + this.deviceMessageService = deviceMessageService;
  33 + this.deviceService = deviceService;
  34 + this.sessionManager = sessionManager;
  35 + this.protocol = protocol;
  36 + }
  37 +
  38 + /**
  39 + * 处理下行消息
  40 + *
  41 + * @param message 下行消息
  42 + */
  43 + public void handle(IotDeviceMessage message) {
  44 + try {
  45 + log.info("[handle][处理下行消息，设备 ID: {}，方法: {}，消息 ID: {}]",
  46 + message.getDeviceId(), message.getMethod(), message.getId());
  47 +
  48 + // 1.1 获取设备信息
  49 + IotDeviceRespDTO deviceInfo = deviceService.getDeviceFromCache(message.getDeviceId());
  50 + if (deviceInfo == null) {
  51 + log.error("[handle][设备不存在，设备 ID: {}]", message.getDeviceId());
  52 + return;
  53 + }
  54 + // 1.2 检查设备是否在线（即是否有地址映射）
  55 + if (sessionManager.isDeviceOffline(message.getDeviceId())) {
  56 + log.warn("[handle][设备不在线，设备 ID: {}]", message.getDeviceId());
  57 + return;
  58 + }
  59 +
  60 + // 2. 根据产品 Key 和设备名称编码消息，并发送到设备
  61 + byte[] bytes = deviceMessageService.encodeDeviceMessage(message, deviceInfo.getProductKey(),
  62 + deviceInfo.getDeviceName());
  63 + DatagramSocket socket = protocol.getUdpSocket();
  64 + if (socket == null) {
  65 + log.error("[handle][UDP Socket 不可用，设备 ID: {}]", message.getDeviceId());
  66 + return;
  67 + }
  68 + boolean success = sessionManager.sendToDevice(message.getDeviceId(), bytes, socket);
  69 + if (success) {
  70 + log.info("[handle][下行消息发送成功，设备 ID: {}，方法: {}，消息 ID: {}，数据长度: {} 字节]",
  71 + message.getDeviceId(), message.getMethod(), message.getId(), bytes.length);
  72 + } else {
  73 + log.error("[handle][下行消息发送失败，设备 ID: {}，方法: {}，消息 ID: {}]",
  74 + message.getDeviceId(), message.getMethod(), message.getId());
  75 + }
  76 + } catch (Exception e) {
  77 + log.error("[handle][处理下行消息失败，设备 ID: {}，方法: {}，消息内容: {}]",
  78 + message.getDeviceId(), message.getMethod(), message, e);
  79 + }
  80 + }
  81 +
  82 +}
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/udp/router/IotUdpUpstreamHandler.java 0 → 100644
  View file @eb5fa9f
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.udp.router;
  2 +
  3 +import cn.hutool.core.map.MapUtil;
  4 +import cn.hutool.core.util.BooleanUtil;
  5 +import cn.hutool.core.util.StrUtil;
  6 +import cn.hutool.extra.spring.SpringUtil;
  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.biz.dto.IotDeviceRespDTO;
  12 +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
  13 +import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils;
  14 +import cn.iocoder.yudao.module.iot.gateway.codec.tcp.IotTcpBinaryDeviceMessageCodec;
  15 +import cn.iocoder.yudao.module.iot.gateway.codec.tcp.IotTcpJsonDeviceMessageCodec;
  16 +import cn.iocoder.yudao.module.iot.gateway.protocol.udp.IotUdpUpstreamProtocol;
  17 +import cn.iocoder.yudao.module.iot.gateway.protocol.udp.manager.IotUdpSessionManager;
  18 +import cn.iocoder.yudao.module.iot.gateway.service.auth.IotDeviceTokenService;
  19 +import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceService;
  20 +import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
  21 +import io.vertx.core.buffer.Buffer;
  22 +import io.vertx.core.datagram.DatagramPacket;
  23 +import io.vertx.core.datagram.DatagramSocket;
  24 +import lombok.extern.slf4j.Slf4j;
  25 +
  26 +import java.net.InetSocketAddress;
  27 +import java.util.Map;
  28 +
  29 +/**
  30 + * UDP 上行消息处理器
  31 + * <p>
  32 + * 采用无状态 Token 机制（每次请求携带 token）：
  33 + * 1. 认证请求：设备发送 auth 消息，携带 clientId、username、password
  34 + * 2. 返回 Token：服务端验证后返回 JWT token
  35 + * 3. 后续请求：每次请求在 params 中携带 token
  36 + * 4. 服务端验证：每次请求通过 IotDeviceTokenService.verifyToken() 验证
  37 + *
  38 + * @author 芋道源码
  39 + */
  40 +@Slf4j
  41 +public class IotUdpUpstreamHandler {
  42 +
  43 + private static final String CODEC_TYPE_JSON = IotTcpJsonDeviceMessageCodec.TYPE;
  44 + private static final String CODEC_TYPE_BINARY = IotTcpBinaryDeviceMessageCodec.TYPE;
  45 +
  46 + private static final String AUTH_METHOD = "auth";
  47 + /**
  48 + * Token 参数 Key
  49 + */
  50 + private static final String PARAM_KEY_TOKEN = "token";
  51 +
  52 + private final IotDeviceMessageService deviceMessageService;
  53 +
  54 + private final IotDeviceService deviceService;
  55 +
  56 + private final IotUdpSessionManager sessionManager;
  57 +
  58 + private final IotDeviceTokenService deviceTokenService;
  59 +
  60 + private final IotDeviceCommonApi deviceApi;
  61 +
  62 + private final String serverId;
  63 +
  64 + public IotUdpUpstreamHandler(IotUdpUpstreamProtocol protocol,
  65 + IotDeviceMessageService deviceMessageService,
  66 + IotDeviceService deviceService,
  67 + IotUdpSessionManager sessionManager) {
  68 + this.deviceMessageService = deviceMessageService;
  69 + this.deviceService = deviceService;
  70 + this.sessionManager = sessionManager;
  71 + this.deviceTokenService = SpringUtil.getBean(IotDeviceTokenService.class);
  72 + this.deviceApi = SpringUtil.getBean(IotDeviceCommonApi.class);
  73 + this.serverId = protocol.getServerId();
  74 + }
  75 +
  76 + /**
  77 + * 处理 UDP 数据包
  78 + *
  79 + * @param packet 数据包
  80 + * @param socket UDP Socket
  81 + */
  82 + public void handle(DatagramPacket packet, DatagramSocket socket) {
  83 + InetSocketAddress senderAddress = new InetSocketAddress(packet.sender().host(), packet.sender().port());
  84 + Buffer data = packet.data();
  85 + log.debug("[handle][收到 UDP 数据包，来源: {}，数据长度: {} 字节]",
  86 + sessionManager.buildAddressKey(senderAddress), data.length());
  87 + try {
  88 + processMessage(data, senderAddress, socket);
  89 + } catch (Exception e) {
  90 + log.error("[handle][处理消息失败，来源: {}，错误: {}]",
  91 + sessionManager.buildAddressKey(senderAddress), e.getMessage(), e);
  92 + // UDP 无连接，不需要断开连接，只记录错误
  93 + }
  94 + }
  95 +
  96 + /**
  97 + * 处理消息
  98 + *
  99 + * @param buffer 消息
  100 + * @param senderAddress 发送者地址
  101 + * @param socket UDP Socket
  102 + */
  103 + private void processMessage(Buffer buffer, InetSocketAddress senderAddress, DatagramSocket socket) {
  104 + // 1. 基础检查
  105 + if (buffer == null || buffer.length() == 0) {
  106 + return;
  107 + }
  108 +
  109 + // 2. 获取消息格式类型
  110 + String codecType = getMessageCodecType(buffer);
  111 +
  112 + // 3. 解码消息
  113 + IotDeviceMessage message;
  114 + try {
  115 + message = deviceMessageService.decodeDeviceMessage(buffer.getBytes(), codecType);
  116 + if (message == null) {
  117 + log.warn("[processMessage][消息解码失败，来源: {}]", sessionManager.buildAddressKey(senderAddress));
  118 + sendErrorResponse(socket, senderAddress, null, "消息解码失败", codecType);
  119 + return;
  120 + }
  121 + } catch (Exception e) {
  122 + log.error("[processMessage][消息解码异常，来源: {}]", sessionManager.buildAddressKey(senderAddress), e);
  123 + sendErrorResponse(socket, senderAddress, null, "消息解码失败: " + e.getMessage(), codecType);
  124 + return;
  125 + }
  126 +
  127 + // 4. 根据消息类型路由处理
  128 + try {
  129 + if (AUTH_METHOD.equals(message.getMethod())) {
  130 + // 认证请求
  131 + handleAuthenticationRequest(message, codecType, senderAddress, socket);
  132 + } else {
  133 + // 业务消息
  134 + handleBusinessRequest(message, codecType, senderAddress, socket);
  135 + }
  136 + } catch (Exception e) {
  137 + log.error("[processMessage][处理消息失败，来源: {}，消息方法: {}]",
  138 + sessionManager.buildAddressKey(senderAddress), message.getMethod(), e);
  139 + sendErrorResponse(socket, senderAddress, message.getRequestId(), "消息处理失败", codecType);
  140 + }
  141 + }
  142 +
  143 + /**
  144 + * 处理认证请求
  145 + *
  146 + * @param message 消息信息
  147 + * @param codecType 消息编解码类型
  148 + * @param senderAddress 发送者地址
  149 + * @param socket UDP Socket
  150 + */
  151 + private void handleAuthenticationRequest(IotDeviceMessage message, String codecType,
  152 + InetSocketAddress senderAddress, DatagramSocket socket) {
  153 + String addressKey = sessionManager.buildAddressKey(senderAddress);
  154 + try {
  155 + // 1.1 解析认证参数
  156 + IotDeviceAuthReqDTO authParams = parseAuthParams(message.getParams());
  157 + if (authParams == null) {
  158 + log.warn("[handleAuthenticationRequest][认证参数解析失败，来源: {}]", addressKey);
  159 + sendErrorResponse(socket, senderAddress, message.getRequestId(), "认证参数不完整", codecType);
  160 + return;
  161 + }
  162 + // 1.2 执行认证
  163 + if (!validateDeviceAuth(authParams)) {
  164 + log.warn("[handleAuthenticationRequest][认证失败，来源: {}，username: {}]",
  165 + addressKey, authParams.getUsername());
  166 + sendErrorResponse(socket, senderAddress, message.getRequestId(), "认证失败", codecType);
  167 + return;
  168 + }
  169 +
  170 + // 2.1 解析设备信息
  171 + IotDeviceAuthUtils.DeviceInfo deviceInfo = IotDeviceAuthUtils.parseUsername(authParams.getUsername());
  172 + if (deviceInfo == null) {
  173 + sendErrorResponse(socket, senderAddress, message.getRequestId(), "解析设备信息失败", codecType);
  174 + return;
  175 + }
  176 + // 2.2 获取设备信息
  177 + IotDeviceRespDTO device = deviceService.getDeviceFromCache(deviceInfo.getProductKey(),
  178 + deviceInfo.getDeviceName());
  179 + if (device == null) {
  180 + sendErrorResponse(socket, senderAddress, message.getRequestId(), "设备不存在", codecType);
  181 + return;
  182 + }
  183 +
  184 + // 3.1 生成 JWT Token（无状态）
  185 + String token = deviceTokenService.createToken(device.getProductKey(), device.getDeviceName());
  186 +
  187 + // 3.2 更新设备地址映射（用于下行消息）
  188 + sessionManager.updateDeviceAddress(device.getId(), senderAddress);
  189 +
  190 + // 3.3 发送上线消息
  191 + sendOnlineMessage(device);
  192 +
  193 + // 3.4 发送成功响应（包含 token）
  194 + sendAuthSuccessResponse(socket, senderAddress, message.getRequestId(), token, codecType);
  195 + log.info("[handleAuthenticationRequest][认证成功，设备 ID: {}，设备名: {}，来源: {}]",
  196 + device.getId(), device.getDeviceName(), addressKey);
  197 + } catch (Exception e) {
  198 + log.error("[handleAuthenticationRequest][认证处理异常，来源: {}]", addressKey, e);
  199 + sendErrorResponse(socket, senderAddress, message.getRequestId(), "认证处理异常", codecType);
  200 + }
  201 + }
  202 +
  203 + /**
  204 + * 处理业务请求
  205 + *
  206 + * @param message 消息信息
  207 + * @param codecType 消息编解码类型
  208 + * @param senderAddress 发送者地址
  209 + * @param socket UDP Socket
  210 + */
  211 + @SuppressWarnings("unchecked")
  212 + private void handleBusinessRequest(IotDeviceMessage message, String codecType,
  213 + InetSocketAddress senderAddress, DatagramSocket socket) {
  214 + String addressKey = sessionManager.buildAddressKey(senderAddress);
  215 + try {
  216 + // 1.1 从消息中提取 token（无状态：消息体携带 token）
  217 + String token = null;
  218 + if (message.getParams() instanceof Map) {
  219 + Map<String, Object> paramsMap = (Map<String, Object>) message.getParams();
  220 + token = (String) paramsMap.remove(PARAM_KEY_TOKEN);
  221 + }
  222 + if (StrUtil.isBlank(token)) {
  223 + log.warn("[handleBusinessRequest][缺少 token，来源: {}]", addressKey);
  224 + sendErrorResponse(socket, senderAddress, message.getRequestId(), "请先进行认证", codecType);
  225 + return;
  226 + }
  227 + // 1.2 验证 token，获取设备信息
  228 + IotDeviceAuthUtils.DeviceInfo deviceInfo = deviceTokenService.verifyToken(token);
  229 + if (deviceInfo == null) {
  230 + log.warn("[handleBusinessRequest][token 无效或已过期，来源: {}]", addressKey);
  231 + sendErrorResponse(socket, senderAddress, message.getRequestId(), "token 无效或已过期", codecType);
  232 + return;
  233 + }
  234 +
  235 + // 2. 获取设备详细信息
  236 + IotDeviceRespDTO device = deviceService.getDeviceFromCache(deviceInfo.getProductKey(),
  237 + deviceInfo.getDeviceName());
  238 + if (device == null) {
  239 + log.warn("[handleBusinessRequest][设备不存在，来源: {}，productKey: {}，deviceName: {}]",
  240 + addressKey, deviceInfo.getProductKey(), deviceInfo.getDeviceName());
  241 + sendErrorResponse(socket, senderAddress, message.getRequestId(), "设备不存在", codecType);
  242 + return;
  243 + }
  244 +
  245 + // 3. 更新设备地址映射（保持最新）
  246 + sessionManager.updateDeviceAddress(device.getId(), senderAddress);
  247 +
  248 + // 4. 发送消息到消息总线
  249 + deviceMessageService.sendDeviceMessage(message, device.getProductKey(),
  250 + device.getDeviceName(), serverId);
  251 +
  252 + // 5. 发送成功响应
  253 + sendSuccessResponse(socket, senderAddress, message.getRequestId(), "处理成功", codecType);
  254 + log.debug("[handleBusinessRequest][业务消息处理成功，设备 ID: {}，方法: {}，来源: {}]",
  255 + device.getId(), message.getMethod(), addressKey);
  256 + } catch (Exception e) {
  257 + log.error("[handleBusinessRequest][业务请求处理异常，来源: {}]", addressKey, e);
  258 + sendErrorResponse(socket, senderAddress, message.getRequestId(), "处理失败", codecType);
  259 + }
  260 + }
  261 +
  262 + /**
  263 + * 获取消息编解码类型
  264 + *
  265 + * @param buffer 消息
  266 + * @return 消息编解码类型
  267 + */
  268 + private String getMessageCodecType(Buffer buffer) {
  269 + // 检测消息格式类型
  270 + return IotTcpBinaryDeviceMessageCodec.isBinaryFormatQuick(buffer.getBytes()) ? CODEC_TYPE_BINARY
  271 + : CODEC_TYPE_JSON;
  272 + }
  273 +
  274 + /**
  275 + * 发送设备上线消息
  276 + *
  277 + * @param device 设备信息
  278 + */
  279 + private void sendOnlineMessage(IotDeviceRespDTO device) {
  280 + try {
  281 + IotDeviceMessage onlineMessage = IotDeviceMessage.buildStateUpdateOnline();
  282 + deviceMessageService.sendDeviceMessage(onlineMessage, device.getProductKey(),
  283 + device.getDeviceName(), serverId);
  284 + } catch (Exception e) {
  285 + log.error("[sendOnlineMessage][发送上线消息失败，设备: {}]", device.getDeviceName(), e);
  286 + }
  287 + }
  288 +
  289 + /**
  290 + * 验证设备认证信息
  291 + *
  292 + * @param authParams 认证参数
  293 + * @return 是否认证成功
  294 + */
  295 + private boolean validateDeviceAuth(IotDeviceAuthReqDTO authParams) {
  296 + try {
  297 + CommonResult<Boolean> result = deviceApi.authDevice(new IotDeviceAuthReqDTO()
  298 + .setClientId(authParams.getClientId()).setUsername(authParams.getUsername())
  299 + .setPassword(authParams.getPassword()));
  300 + result.checkError();
  301 + return BooleanUtil.isTrue(result.getData());
  302 + } catch (Exception e) {
  303 + log.error("[validateDeviceAuth][设备认证异常，username: {}]", authParams.getUsername(), e);
  304 + return false;
  305 + }
  306 + }
  307 +
  308 + /**
  309 + * 发送认证成功响应（包含 token）
  310 + *
  311 + * @param socket UDP Socket
  312 + * @param address 目标地址
  313 + * @param requestId 请求 ID
  314 + * @param token JWT Token
  315 + * @param codecType 消息编解码类型
  316 + */
  317 + private void sendAuthSuccessResponse(DatagramSocket socket, InetSocketAddress address,
  318 + String requestId, String token, String codecType) {
  319 + try {
  320 + Object responseData = MapUtil.builder()
  321 + .put("success", true)
  322 + .put("token", token)
  323 + .put("message", "认证成功")
  324 + .build();
  325 +
  326 + IotDeviceMessage responseMessage = IotDeviceMessage.replyOf(requestId, AUTH_METHOD, responseData, 0, "认证成功");
  327 + byte[] encodedData = deviceMessageService.encodeDeviceMessage(responseMessage, codecType);
  328 + socket.send(Buffer.buffer(encodedData), address.getPort(), address.getHostString(), result -> {
  329 + if (result.failed()) {
  330 + log.error("[sendAuthSuccessResponse][发送认证成功响应失败，地址: {}]",
  331 + sessionManager.buildAddressKey(address), result.cause());
  332 + }
  333 + });
  334 + } catch (Exception e) {
  335 + log.error("[sendAuthSuccessResponse][发送认证成功响应异常，地址: {}]",
  336 + sessionManager.buildAddressKey(address), e);
  337 + }
  338 + }
  339 +
  340 + /**
  341 + * 发送成功响应
  342 + *
  343 + * @param socket UDP Socket
  344 + * @param address 目标地址
  345 + * @param requestId 请求 ID
  346 + * @param message 消息
  347 + * @param codecType 消息编解码类型
  348 + */
  349 + @SuppressWarnings("SameParameterValue")
  350 + private void sendSuccessResponse(DatagramSocket socket, InetSocketAddress address,
  351 + String requestId, String message, String codecType) {
  352 + sendResponse(socket, address, true, message, requestId, codecType);
  353 + }
  354 +
  355 + /**
  356 + * 发送错误响应
  357 + *
  358 + * @param socket UDP Socket
  359 + * @param address 目标地址
  360 + * @param requestId 请求 ID
  361 + * @param errorMessage 错误消息
  362 + * @param codecType 消息编解码类型
  363 + */
  364 + private void sendErrorResponse(DatagramSocket socket, InetSocketAddress address,
  365 + String requestId, String errorMessage, String codecType) {
  366 + sendResponse(socket, address, false, errorMessage, requestId, codecType);
  367 + }
  368 +
  369 + /**
  370 + * 发送响应消息
  371 + *
  372 + * @param socket UDP Socket
  373 + * @param address 目标地址
  374 + * @param success 是否成功
  375 + * @param message 消息
  376 + * @param requestId 请求 ID
  377 + * @param codecType 消息编解码类型
  378 + */
  379 + private void sendResponse(DatagramSocket socket, InetSocketAddress address, boolean success,
  380 + String message, String requestId, String codecType) {
  381 + try {
  382 + Object responseData = MapUtil.builder()
  383 + .put("success", success)
  384 + .put("message", message)
  385 + .build();
  386 +
  387 + int code = success ? 0 : 401;
  388 + IotDeviceMessage responseMessage = IotDeviceMessage.replyOf(requestId, "response", responseData,
  389 + code, message);
  390 +
  391 + byte[] encodedData = deviceMessageService.encodeDeviceMessage(responseMessage, codecType);
  392 + socket.send(Buffer.buffer(encodedData), address.getPort(), address.getHostString(), ar -> {
  393 + if (ar.failed()) {
  394 + log.error("[sendResponse][发送响应失败，地址: {}]",
  395 + sessionManager.buildAddressKey(address), ar.cause());
  396 + }
  397 + });
  398 + } catch (Exception e) {
  399 + log.error("[sendResponse][发送响应异常，地址: {}]",
  400 + sessionManager.buildAddressKey(address), e);
  401 + }
  402 + }
  403 +
  404 + /**
  405 + * 解析认证参数
  406 + *
  407 + * @param params 参数对象（通常为 Map 类型）
  408 + * @return 认证参数 DTO，解析失败时返回 null
  409 + */
  410 + @SuppressWarnings("unchecked")
  411 + private IotDeviceAuthReqDTO parseAuthParams(Object params) {
  412 + if (params == null) {
  413 + return null;
  414 + }
  415 +
  416 + try {
  417 + // 参数默认为 Map 类型，直接转换
  418 + if (params instanceof Map) {
  419 + Map<String, Object> paramMap = (Map<String, Object>) params;
  420 + return new IotDeviceAuthReqDTO()
  421 + .setClientId(MapUtil.getStr(paramMap, "clientId"))
  422 + .setUsername(MapUtil.getStr(paramMap, "username"))
  423 + .setPassword(MapUtil.getStr(paramMap, "password"));
  424 + }
  425 +
  426 + // 如果已经是目标类型，直接返回
  427 + if (params instanceof IotDeviceAuthReqDTO) {
  428 + return (IotDeviceAuthReqDTO) params;
  429 + }
  430 +
  431 + // 其他情况尝试 JSON 转换
  432 + String jsonStr = JsonUtils.toJsonString(params);
  433 + return JsonUtils.parseObject(jsonStr, IotDeviceAuthReqDTO.class);
  434 + } catch (Exception e) {
  435 + log.error("[parseAuthParams][解析认证参数({})失败]", params, e);
  436 + return null;
  437 + }
  438 + }
  439 +
  440 +}
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application.yaml
  View file @eb5fa9f
... ... @@ -96,6 +96,16 @@ yudao:
96 96 ssl-cert-path: "classpath:certs/client.jks"
97 97 ssl-key-path: "classpath:certs/client.jks"
98 98 # ====================================
  99 + # 针对引入的 UDP 组件的配置
  100 + # ====================================
  101 + udp:
  102 + enabled: false # 是否启用 UDP
  103 + port: 8092 # UDP 服务端口
  104 + receive-buffer-size: 65536 # 接收缓冲区大小（字节，默认 64KB）
  105 + send-buffer-size: 65536 # 发送缓冲区大小（字节，默认 64KB）
  106 + session-timeout-ms: 60000 # 会话超时时间（毫秒，默认 60 秒）
  107 + session-clean-interval-ms: 30000 # 会话清理间隔（毫秒，默认 30 秒）
  108 + # ====================================
99 109 # 针对引入的 MQTT 组件的配置
100 110 # ====================================
101 111 mqtt:
... ...
yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/udp/IotUdpProtocolIntegrationTest.java 0 → 100644
  View file @eb5fa9f
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.udp;
  2 +
  3 +import cn.hutool.core.map.MapUtil;
  4 +import cn.hutool.core.util.IdUtil;
  5 +import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
  6 +import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
  7 +import lombok.extern.slf4j.Slf4j;
  8 +import org.junit.jupiter.api.Test;
  9 +
  10 +import java.net.DatagramPacket;
  11 +import java.net.DatagramSocket;
  12 +import java.net.InetAddress;
  13 +import java.nio.charset.StandardCharsets;
  14 +
  15 +/**
  16 + * IoT 网关 UDP 协议集成测试（手动测试）
  17 + *
  18 + * <p>使用步骤：
  19 + * <ol>
  20 + * <li>启动 yudao-module-iot-gateway 服务（UDP 端口 8092）</li>
  21 + * <li>运行 {@link #testAuth()} 获取 token，将返回的 token 粘贴到 {@link #TOKEN} 常量</li>
  22 + * <li>运行 {@link #testPropertyPost()} 测试属性上报，或运行 {@link #testEventPost()} 测试事件上报</li>
  23 + * </ol>
  24 + *
  25 + * @author 芋道源码
  26 + */
  27 +@Slf4j
  28 +public class IotUdpProtocolIntegrationTest {
  29 +
  30 + private static final String SERVER_HOST = "127.0.0.1";
  31 + private static final int SERVER_PORT = 8092;
  32 + private static final int TIMEOUT_MS = 5000;
  33 +
  34 + // 设备信息（根据实际情况修改 PRODUCT_KEY、DEVICE_NAME、PASSWORD）
  35 + private static final String PRODUCT_KEY = "4aymZgOTOOCrDKRT";
  36 + private static final String DEVICE_NAME = "small";
  37 + private static final String PASSWORD = "509e2b08f7598eb139d276388c600435913ba4c94cd0d50aebc5c0d1855bcb75";
  38 +
  39 + // TODO @芋艿：1、IotDeviceAuthUtils 调整下拼接；2、password 的生成；3、后续给 http 也整个单测；4、后续给 tcp 也整个单测；5、后续给 mqtt 也整个单测；6、后续给 emqp 也整个单测
  40 + private static final String CLIENT_ID = PRODUCT_KEY + "." + DEVICE_NAME;
  41 + private static final String USERNAME = DEVICE_NAME + "&" + PRODUCT_KEY;
  42 +
  43 + /**
  44 + * 设备 Token：从 {@link #testAuth()} 方法获取后，粘贴到这里
  45 + */
  46 + private static final String TOKEN = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJwcm9kdWN0S2V5IjoiNGF5bVpnT1RPT0NyREtSVCIsImV4cCI6MTc2OTMxMTY0NiwiZGV2aWNlTmFtZSI6InNtYWxsIn0.re6LCaRfKiE9VQTP3w0Brh2ScVIgrvN3H96z_snndoM";
  47 +
  48 + /**
  49 + * 认证测试：获取设备 Token
  50 + */
  51 + @Test
  52 + public void testAuth() throws Exception {
  53 + String payload = JsonUtils.toJsonString(MapUtil.builder()
  54 + .put("id", IdUtil.fastSimpleUUID())
  55 + .put("method", "auth")
  56 + .put("params", MapUtil.builder()
  57 + .put("clientId", CLIENT_ID)
  58 + .put("username", USERNAME)
  59 + .put("password", PASSWORD)
  60 + .build())
  61 + .build());
  62 +
  63 + try (DatagramSocket socket = new DatagramSocket()) {
  64 + socket.setSoTimeout(TIMEOUT_MS);
  65 +
  66 + log.info("[testAuth][请求体: {}]", payload);
  67 +
  68 + String response = sendAndReceive(socket, payload);
  69 +
  70 + log.info("[testAuth][响应体: {}]", response);
  71 + log.info("[testAuth][请将返回的 token 复制到 TOKEN 常量中]");
  72 + }
  73 + }
  74 +
  75 + /**
  76 + * 属性上报测试
  77 + */
  78 + @Test
  79 + public void testPropertyPost() throws Exception {
  80 + String payload = JsonUtils.toJsonString(MapUtil.builder()
  81 + .put("id", IdUtil.fastSimpleUUID())
  82 + .put("method", IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod())
  83 + .put("version", "1.0")
  84 + .put("params", MapUtil.builder()
  85 + .put("token", TOKEN)
  86 + .put("width", 1)
  87 + .put("height", "2")
  88 + .build())
  89 + .build());
  90 +
  91 + try (DatagramSocket socket = new DatagramSocket()) {
  92 + socket.setSoTimeout(TIMEOUT_MS);
  93 +
  94 + log.info("[testPropertyPost][请求体: {}]", payload);
  95 +
  96 + String response = sendAndReceive(socket, payload);
  97 +
  98 + log.info("[testPropertyPost][响应体: {}]", response);
  99 + }
  100 + }
  101 +
  102 + /**
  103 + * 事件上报测试
  104 + */
  105 + @Test
  106 + public void testEventPost() throws Exception {
  107 + String payload = JsonUtils.toJsonString(MapUtil.builder()
  108 + .put("id", IdUtil.fastSimpleUUID())
  109 + .put("method", IotDeviceMessageMethodEnum.EVENT_POST.getMethod())
  110 + .put("version", "1.0")
  111 + .put("params", MapUtil.builder()
  112 + .put("identifier", "eat")
  113 + .put("value", MapUtil.builder()
  114 + .put("width", 1)
  115 + .put("height", "2")
  116 + .put("oneThree", "3")
  117 + .build())
  118 + .put("time", System.currentTimeMillis())
  119 + .build())
  120 + .build());
  121 +
  122 + try (DatagramSocket socket = new DatagramSocket()) {
  123 + socket.setSoTimeout(TIMEOUT_MS);
  124 +
  125 + log.info("[testEventPost][请求体: {}]", payload);
  126 +
  127 + String response = sendAndReceive(socket, payload);
  128 +
  129 + log.info("[testEventPost][响应体: {}]", response);
  130 + }
  131 + }
  132 +
  133 + /**
  134 + * 发送 UDP 请求并接收响应
  135 + *
  136 + * @param socket UDP Socket
  137 + * @param payload 请求体
  138 + * @return 响应内容
  139 + */
  140 + private String sendAndReceive(DatagramSocket socket, String payload) throws Exception {
  141 + byte[] sendData = payload.getBytes(StandardCharsets.UTF_8);
  142 + InetAddress address = InetAddress.getByName(SERVER_HOST);
  143 +
  144 + // 发送请求
  145 + DatagramPacket sendPacket = new DatagramPacket(sendData, sendData.length, address, SERVER_PORT);
  146 + socket.send(sendPacket);
  147 +
  148 + // 接收响应
  149 + byte[] receiveData = new byte[4096];
  150 + DatagramPacket receivePacket = new DatagramPacket(receiveData, receiveData.length);
  151 + try {
  152 + socket.receive(receivePacket);
  153 + return new String(receivePacket.getData(), 0, receivePacket.getLength(), StandardCharsets.UTF_8);
  154 + } catch (java.net.SocketTimeoutException e) {
  155 + log.warn("[sendAndReceive][接收响应超时]");
  156 + return null;
  157 + }
  158 + }
  159 +
  160 +}
... ...