package cn.iocoder.yudao.module.iot.gateway.protocol.udp.handler.upstream; import cn.hutool.core.map.MapUtil; import cn.hutool.core.util.ArrayUtil; import cn.hutool.core.util.BooleanUtil; import cn.hutool.core.util.IdUtil; 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.biz.IotDeviceCommonApi; import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceAuthReqDTO; import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceRespDTO; import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum; import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; import cn.iocoder.yudao.module.iot.core.topic.IotDeviceIdentity; import cn.iocoder.yudao.module.iot.core.topic.auth.IotDeviceRegisterReqDTO; import cn.iocoder.yudao.module.iot.core.topic.auth.IotDeviceRegisterRespDTO; import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils; import cn.iocoder.yudao.module.iot.gateway.protocol.udp.manager.IotUdpSessionManager; import cn.iocoder.yudao.module.iot.gateway.serialize.IotMessageSerializer; import cn.iocoder.yudao.module.iot.gateway.service.auth.IotDeviceTokenService; 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.buffer.Buffer; import io.vertx.core.datagram.DatagramPacket; import io.vertx.core.datagram.DatagramSocket; import lombok.extern.slf4j.Slf4j; import cn.hutool.core.lang.Assert; import java.net.InetSocketAddress; import java.util.Map; import static cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants.*; import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception; import static cn.iocoder.yudao.module.iot.gateway.enums.ErrorCodeConstants.DEVICE_AUTH_FAIL; /** * UDP 上行消息处理器 *

* 采用无状态 Token 机制（每次请求携带 token）： * 1. 认证请求：设备发送 auth 消息，携带 clientId、username、password * 2. 返回 Token：服务端验证后返回 JWT token * 3. 后续请求：每次请求在 params 中携带 token * 4. 服务端验证：每次请求通过 IotDeviceTokenService.verifyToken() 验证 * * @author 芋道源码 */ @Slf4j public class IotUdpUpstreamHandler { private static final String AUTH_METHOD = "auth"; /** * Token 参数 Key */ private static final String PARAM_KEY_TOKEN = "token"; /** * Body 参数 Key（实际请求内容） */ private static final String PARAM_KEY_BODY = "body"; private final String serverId; /** * 消息序列化器（处理业务消息序列化/反序列化） */ private final IotMessageSerializer serializer; /** * UDP 会话管理器 */ private final IotUdpSessionManager sessionManager; private final IotDeviceMessageService deviceMessageService; private final IotDeviceService deviceService; private final IotDeviceTokenService deviceTokenService; private final IotDeviceCommonApi deviceApi; public IotUdpUpstreamHandler(String serverId, IotUdpSessionManager sessionManager, IotMessageSerializer serializer) { Assert.notNull(serializer, "消息序列化器必须配置"); Assert.notNull(sessionManager, "会话管理器不能为空"); this.serverId = serverId; this.sessionManager = sessionManager; this.serializer = serializer; this.deviceMessageService = SpringUtil.getBean(IotDeviceMessageService.class); this.deviceService = SpringUtil.getBean(IotDeviceService.class); this.deviceApi = SpringUtil.getBean(IotDeviceCommonApi.class); this.deviceTokenService = SpringUtil.getBean(IotDeviceTokenService.class); } /** * 处理 UDP 数据包 * * @param packet 数据包 * @param socket UDP Socket */ public void handle(DatagramPacket packet, DatagramSocket socket) { InetSocketAddress senderAddress = new InetSocketAddress(packet.sender().host(), packet.sender().port()); Buffer data = packet.data(); String addressKey = sessionManager.buildAddressKey(senderAddress); log.debug("[handle][收到 UDP 数据包，来源: {}，数据长度: {} 字节]", addressKey, data.length()); processMessage(data, senderAddress, socket); } /** * 处理消息 * * @param buffer 消息 * @param senderAddress 发送者地址 * @param socket UDP Socket */ private void processMessage(Buffer buffer, InetSocketAddress senderAddress, DatagramSocket socket) { String addressKey = sessionManager.buildAddressKey(senderAddress); // 1.1 基础检查 if (ArrayUtil.isEmpty(buffer)) { return; } // 1.2 反序列化消息 IotDeviceMessage message = serializer.deserialize(buffer.getBytes()); if (message == null) { sendErrorResponse(socket, senderAddress, null, null, BAD_REQUEST.getCode(), "消息反序列化失败"); return; } // 2. 根据消息类型路由处理 try { if (AUTH_METHOD.equals(message.getMethod())) { // 认证请求 handleAuthenticationRequest(message, senderAddress, socket); } else if (IotDeviceMessageMethodEnum.DEVICE_REGISTER.getMethod().equals(message.getMethod())) { // 设备动态注册请求 handleRegisterRequest(message, senderAddress, socket); } else { // 业务消息 handleBusinessRequest(message, senderAddress, socket); } } catch (ServiceException e) { // 业务异常，返回对应的错误码和错误信息 log.warn("[processMessage][业务异常，来源: {}，requestId: {}，method: {}，错误: {}]", addressKey, message.getRequestId(), message.getMethod(), e.getMessage()); sendErrorResponse(socket, senderAddress, message.getRequestId(), message.getMethod(), e.getCode(), e.getMessage()); } catch (IllegalArgumentException e) { // 参数校验失败，返回 400 log.warn("[processMessage][参数校验失败，来源: {}，requestId: {}，method: {}，错误: {}]", addressKey, message.getRequestId(), message.getMethod(), e.getMessage()); sendErrorResponse(socket, senderAddress, message.getRequestId(), message.getMethod(), BAD_REQUEST.getCode(), e.getMessage()); } catch (Exception e) { // 其他异常，返回 500 log.error("[processMessage][处理消息失败，来源: {}，requestId: {}，method: {}]", addressKey, message.getRequestId(), message.getMethod(), e); sendErrorResponse(socket, senderAddress, message.getRequestId(), message.getMethod(), INTERNAL_SERVER_ERROR.getCode(), INTERNAL_SERVER_ERROR.getMsg()); } } /** * 处理认证请求 * * @param message 消息信息 * @param senderAddress 发送者地址 * @param socket UDP Socket */ @SuppressWarnings("DuplicatedCode") private void handleAuthenticationRequest(IotDeviceMessage message, InetSocketAddress senderAddress, DatagramSocket socket) { String clientId = IdUtil.simpleUUID(); // 1. 解析认证参数 IotDeviceAuthReqDTO authParams = JsonUtils.convertObject(message.getParams(), IotDeviceAuthReqDTO.class); Assert.notNull(authParams, "认证参数不能为空"); Assert.notBlank(authParams.getUsername(), "username 不能为空"); Assert.notBlank(authParams.getPassword(), "password 不能为空"); // 2.1 执行认证 CommonResult authResult = deviceApi.authDevice(authParams); authResult.checkError(); if (!BooleanUtil.isTrue(authResult.getData())) { throw exception(DEVICE_AUTH_FAIL); } // 2.2 解析设备信息 IotDeviceIdentity deviceInfo = IotDeviceAuthUtils.parseUsername(authParams.getUsername()); Assert.notNull(deviceInfo, "解析设备信息失败"); // 2.3 获取设备信息 IotDeviceRespDTO device = deviceService.getDeviceFromCache(deviceInfo.getProductKey(), deviceInfo.getDeviceName()); Assert.notNull(device, "设备不存在"); // 3. 生成 JWT Token（无状态） String token = deviceTokenService.createToken(device.getProductKey(), device.getDeviceName()); // 4.1 注册会话 registerSession(senderAddress, device, clientId); // 4.2 发送上线消息 sendOnlineMessage(device); // 4.3 发送成功响应（包含 token） sendSuccessResponse(socket, senderAddress, message.getRequestId(), AUTH_METHOD, MapUtil.of("token", token)); log.info("[handleAuthenticationRequest][认证成功，设备 ID: {}，设备名: {}，来源: {}]", device.getId(), device.getDeviceName(), sessionManager.buildAddressKey(senderAddress)); } /** * 处理设备动态注册请求（一型一密，不需要认证） * * @param message 消息信息 * @param senderAddress 发送者地址 * @param socket UDP Socket * @see 阿里云 - 一型一密 */ @SuppressWarnings("DuplicatedCode") private void handleRegisterRequest(IotDeviceMessage message, InetSocketAddress senderAddress, DatagramSocket socket) { // 1. 解析注册参数 IotDeviceRegisterReqDTO params = JsonUtils.convertObject(message.getParams(), IotDeviceRegisterReqDTO.class); Assert.notNull(params, "注册参数不能为空"); Assert.notBlank(params.getProductKey(), "productKey 不能为空"); Assert.notBlank(params.getDeviceName(), "deviceName 不能为空"); Assert.notBlank(params.getSign(), "sign 不能为空"); // 2. 调用动态注册 CommonResult result = deviceApi.registerDevice(params); result.checkError(); // 3. 发送成功响应 sendSuccessResponse(socket, senderAddress, message.getRequestId(), IotDeviceMessageMethodEnum.DEVICE_REGISTER.getMethod(), result.getData()); log.info("[handleRegisterRequest][注册成功，来源: {}，设备名: {}]", sessionManager.buildAddressKey(senderAddress), params.getDeviceName()); } /** * 处理业务请求 *

* 请求参数格式： * - token：JWT 令牌 * - body：实际请求内容（可以是 Map、List 或其他类型） * * @param message 消息信息 * @param senderAddress 发送者地址 * @param socket UDP Socket */ @SuppressWarnings("unchecked") private void handleBusinessRequest(IotDeviceMessage message, InetSocketAddress senderAddress, DatagramSocket socket) { String addressKey = sessionManager.buildAddressKey(senderAddress); // 1.1 从消息中提取 token 和 body String token = null; Object body = null; if (message.getParams() instanceof Map) { Map paramsMap = (Map) message.getParams(); token = (String) paramsMap.get(PARAM_KEY_TOKEN); body = paramsMap.get(PARAM_KEY_BODY); } if (StrUtil.isBlank(token)) { log.warn("[handleBusinessRequest][缺少 token，来源: {}]", addressKey); sendErrorResponse(socket, senderAddress, message.getRequestId(), message.getMethod(), UNAUTHORIZED.getCode(), "请先进行认证"); return; } // 1.2 验证 token，获取设备信息 IotDeviceIdentity deviceInfo = deviceTokenService.verifyToken(token); if (deviceInfo == null) { log.warn("[handleBusinessRequest][token 无效或已过期，来源: {}]", addressKey); sendErrorResponse(socket, senderAddress, message.getRequestId(), message.getMethod(), UNAUTHORIZED.getCode(), "token 无效或已过期"); return; } // 1.3 获取设备详细信息 IotDeviceRespDTO device = deviceService.getDeviceFromCache(deviceInfo.getProductKey(), deviceInfo.getDeviceName()); if (device == null) { log.warn("[handleBusinessRequest][设备不存在，来源: {}，productKey: {}，deviceName: {}]", addressKey, deviceInfo.getProductKey(), deviceInfo.getDeviceName()); sendErrorResponse(socket, senderAddress, message.getRequestId(), message.getMethod(), BAD_REQUEST.getCode(), "设备不存在"); return; } // 2. 更新会话地址（如有变化） sessionManager.updateSessionAddress(device.getId(), senderAddress); // 3. 将 body 设置为实际的 params，发送消息到消息总线 message.setParams(body); deviceMessageService.sendDeviceMessage(message, device.getProductKey(), device.getDeviceName(), serverId); log.debug("[handleBusinessRequest][业务消息处理成功，设备 ID: {}，方法: {}，来源: {}]", device.getId(), message.getMethod(), addressKey); } /** * 注册会话信息 * * @param address 设备地址 * @param device 设备 * @param clientId 客户端 ID */ private void registerSession(InetSocketAddress address, IotDeviceRespDTO device, String clientId) { IotUdpSessionManager.SessionInfo sessionInfo = new IotUdpSessionManager.SessionInfo() .setDeviceId(device.getId()) .setProductKey(device.getProductKey()) .setDeviceName(device.getDeviceName()) .setAddress(address); sessionManager.registerSession(device.getId(), sessionInfo); } /** * 发送设备上线消息 * * @param device 设备信息 */ private void sendOnlineMessage(IotDeviceRespDTO device) { IotDeviceMessage onlineMessage = IotDeviceMessage.buildStateUpdateOnline(); deviceMessageService.sendDeviceMessage(onlineMessage, device.getProductKey(), device.getDeviceName(), serverId); } // ===================== 发送响应消息 ===================== /** * 发送成功响应 * * @param socket UDP Socket * @param address 目标地址 * @param requestId 请求 ID * @param method 方法名 * @param data 响应数据 */ private void sendSuccessResponse(DatagramSocket socket, InetSocketAddress address, String requestId, String method, Object data) { IotDeviceMessage responseMessage = IotDeviceMessage.replyOf(requestId, method, data, SUCCESS.getCode(), null); writeResponse(socket, address, responseMessage); } /** * 发送错误响应 * * @param socket UDP Socket * @param address 目标地址 * @param requestId 请求 ID * @param method 方法名 * @param code 错误码 * @param msg 错误消息 */ private void sendErrorResponse(DatagramSocket socket, InetSocketAddress address, String requestId, String method, Integer code, String msg) { IotDeviceMessage responseMessage = IotDeviceMessage.replyOf(requestId, method, null, code, msg); writeResponse(socket, address, responseMessage); } /** * 写入响应到 Socket * * @param socket UDP Socket * @param address 目标地址 * @param responseMessage 响应消息 */ private void writeResponse(DatagramSocket socket, InetSocketAddress address, IotDeviceMessage responseMessage) { try { byte[] serializedData = serializer.serialize(responseMessage); socket.send(Buffer.buffer(serializedData), address.getPort(), address.getHostString(), result -> { if (result.failed()) { log.error("[writeResponse][发送响应失败，地址: {}]", sessionManager.buildAddressKey(address), result.cause()); } }); } catch (Exception e) { log.error("[writeResponse][发送响应异常，地址: {}]", sessionManager.buildAddressKey(address), e); } } }