package cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpserver.handler.upstream; import cn.hutool.core.lang.Assert; import cn.hutool.core.map.MapUtil; import cn.hutool.core.util.BooleanUtil; import cn.hutool.core.util.ObjUtil; import cn.hutool.core.util.StrUtil; import cn.iocoder.yudao.framework.common.exception.ServiceException; import cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants; 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.biz.dto.IotModbusDeviceConfigRespDTO; import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusPointRespDTO; import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum; import cn.iocoder.yudao.module.iot.core.enums.modbus.IotModbusFrameFormatEnum; 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.util.IotDeviceAuthUtils; import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.utils.IotModbusCommonUtils; import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpserver.codec.IotModbusFrame; import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpserver.codec.IotModbusFrameEncoder; import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpserver.manager.IotModbusTcpServerConfigCacheService; import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpserver.manager.IotModbusTcpServerConnectionManager; import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpserver.manager.IotModbusTcpServerConnectionManager.ConnectionInfo; import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpserver.manager.IotModbusTcpServerPendingRequestManager; import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpserver.manager.IotModbusTcpServerPendingRequestManager.PendingRequest; import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpserver.manager.IotModbusTcpServerPollScheduler; 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.net.NetSocket; import lombok.extern.slf4j.Slf4j; import java.util.Map; import static cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants.BAD_REQUEST; import static cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants.INTERNAL_SERVER_ERROR; import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.invalidParamException; /** * IoT Modbus TCP Server 上行数据处理器 *

* 处理: * 1. 自定义 FC 认证 * 2. 轮询响应 → 点位翻译 → thing.property.post * * @author 芋道源码 */ @Slf4j public class IotModbusTcpServerUpstreamHandler { private static final String METHOD_AUTH = "auth"; private final IotDeviceCommonApi deviceApi; private final IotDeviceMessageService messageService; private final IotModbusFrameEncoder frameEncoder; private final IotModbusTcpServerConnectionManager connectionManager; private final IotModbusTcpServerConfigCacheService configCacheService; private final IotModbusTcpServerPendingRequestManager pendingRequestManager; private final IotModbusTcpServerPollScheduler pollScheduler; private final IotDeviceService deviceService; private final String serverId; public IotModbusTcpServerUpstreamHandler(IotDeviceCommonApi deviceApi, IotDeviceMessageService messageService, IotModbusFrameEncoder frameEncoder, IotModbusTcpServerConnectionManager connectionManager, IotModbusTcpServerConfigCacheService configCacheService, IotModbusTcpServerPendingRequestManager pendingRequestManager, IotModbusTcpServerPollScheduler pollScheduler, IotDeviceService deviceService, String serverId) { this.deviceApi = deviceApi; this.messageService = messageService; this.frameEncoder = frameEncoder; this.connectionManager = connectionManager; this.configCacheService = configCacheService; this.pendingRequestManager = pendingRequestManager; this.pollScheduler = pollScheduler; this.deviceService = deviceService; this.serverId = serverId; } // ========== 帧处理入口 ========== /** * 处理帧 */ public void handleFrame(NetSocket socket, IotModbusFrame frame, IotModbusFrameFormatEnum frameFormat) { if (frame == null) { return; } // 1. 异常响应 if (frame.isException()) { log.warn("[handleFrame][设备异常响应, slaveId={}, FC={}, exceptionCode={}]", frame.getSlaveId(), frame.getFunctionCode(), frame.getExceptionCode()); return; } // 2. 情况一:自定义功能码(认证等扩展) if (StrUtil.isNotEmpty(frame.getCustomData())) { handleCustomFrame(socket, frame, frameFormat); return; } // 3. 情况二:标准 Modbus 响应 → 轮询响应处理 handlePollingResponse(socket, frame, frameFormat); } // ========== 自定义 FC 处理(认证等) ========== /** * 处理自定义功能码帧 *

* 异常分层翻译,参考 {@link cn.iocoder.yudao.module.iot.gateway.protocol.http.handler.upstream.IotHttpAbstractHandler} */ private void handleCustomFrame(NetSocket socket, IotModbusFrame frame, IotModbusFrameFormatEnum frameFormat) { String method = null; try { IotDeviceMessage message = JsonUtils.parseObject(frame.getCustomData(), IotDeviceMessage.class); if (message == null) { throw invalidParamException("自定义 FC 数据解析失败"); } method = message.getMethod(); if (METHOD_AUTH.equals(method)) { handleAuth(socket, frame, frameFormat, message.getParams()); return; } log.warn("[handleCustomFrame][未知 method: {}, frame: slaveId={}, FC={}, customData={}]", method, frame.getSlaveId(), frame.getFunctionCode(), frame.getCustomData()); } catch (ServiceException e) { // 已知业务异常,返回对应的错误码和错误信息 sendCustomResponse(socket, frame, frameFormat, method, e.getCode(), e.getMessage()); } catch (IllegalArgumentException e) { // 参数校验异常,返回 400 错误 sendCustomResponse(socket, frame, frameFormat, method, BAD_REQUEST.getCode(), e.getMessage()); } catch (Exception e) { // 其他未知异常,返回 500 错误 log.error("[handleCustomFrame][解析自定义 FC 数据失败, frame: slaveId={}, FC={}, customData={}]", frame.getSlaveId(), frame.getFunctionCode(), frame.getCustomData(), e); sendCustomResponse(socket, frame, frameFormat, method, INTERNAL_SERVER_ERROR.getCode(), INTERNAL_SERVER_ERROR.getMsg()); } } /** * 处理认证请求 */ @SuppressWarnings("DataFlowIssue") private void handleAuth(NetSocket socket, IotModbusFrame frame, IotModbusFrameFormatEnum frameFormat, Object params) { // 1. 解析认证参数 IotDeviceAuthReqDTO request = JsonUtils.convertObject(params, IotDeviceAuthReqDTO.class); Assert.notNull(request, "认证参数不能为空"); Assert.notBlank(request.getUsername(), "username 不能为空"); Assert.notBlank(request.getPassword(), "password 不能为空"); // 特殊:考虑到 modbus 消息体积较小,默认 clientId 传递空串 if (StrUtil.isBlank(request.getClientId())) { request.setClientId(IotDeviceAuthUtils.buildClientIdFromUsername(request.getUsername())); } Assert.notBlank(request.getClientId(), "clientId 不能为空"); // 2.1 调用认证 API CommonResult result = deviceApi.authDevice(request); result.checkError(); if (BooleanUtil.isFalse(result.getData())) { log.warn("[handleAuth][认证失败, clientId={}, username={}]", request.getClientId(), request.getUsername()); sendCustomResponse(socket, frame, frameFormat, METHOD_AUTH, BAD_REQUEST.getCode(), "认证失败"); return; } // 2.2 解析设备信息 IotDeviceIdentity deviceInfo = IotDeviceAuthUtils.parseUsername(request.getUsername()); Assert.notNull(deviceInfo, "解析设备信息失败"); // 2.3 获取设备信息 IotDeviceRespDTO device = deviceService.getDeviceFromCache(deviceInfo.getProductKey(), deviceInfo.getDeviceName()); Assert.notNull(device, "设备不存在"); // 2.4 加载设备 Modbus 配置,无配置则阻断认证 IotModbusDeviceConfigRespDTO modbusConfig = configCacheService.loadDeviceConfig(device.getId()); if (modbusConfig == null) { log.warn("[handleAuth][设备 {} 没有 Modbus 点位配置, 拒绝认证]", device.getId()); sendCustomResponse(socket, frame, frameFormat, METHOD_AUTH, BAD_REQUEST.getCode(), "设备无 Modbus 配置"); return; } // 2.5 协议不一致,阻断认证 if (ObjUtil.notEqual(frameFormat.getFormat(), modbusConfig.getFrameFormat())) { log.warn("[handleAuth][设备 {} frameFormat 不一致, 连接协议={}, 设备配置={},拒绝认证]", device.getId(), frameFormat.getFormat(), modbusConfig.getFrameFormat()); sendCustomResponse(socket, frame, frameFormat, METHOD_AUTH, BAD_REQUEST.getCode(), "frameFormat 协议不一致"); return; } // 3.1 注册连接 ConnectionInfo connectionInfo = new ConnectionInfo() .setDeviceId(device.getId()) .setProductKey(deviceInfo.getProductKey()) .setDeviceName(deviceInfo.getDeviceName()) .setSlaveId(frame.getSlaveId()) .setFrameFormat(frameFormat); connectionManager.registerConnection(socket, connectionInfo); // 3.2 发送上线消息 IotDeviceMessage onlineMessage = IotDeviceMessage.buildStateUpdateOnline(); messageService.sendDeviceMessage(onlineMessage, deviceInfo.getProductKey(), deviceInfo.getDeviceName(), serverId); // 3.3 发送成功响应 sendCustomResponse(socket, frame, frameFormat, METHOD_AUTH, GlobalErrorCodeConstants.SUCCESS.getCode(), "success"); log.info("[handleAuth][认证成功, clientId={}, deviceId={}]", request.getClientId(), device.getId()); // 4. 启动轮询 pollScheduler.updatePolling(modbusConfig); } /** * 发送自定义功能码响应 */ private void sendCustomResponse(NetSocket socket, IotModbusFrame frame, IotModbusFrameFormatEnum frameFormat, String method, int code, String message) { Map response = MapUtil.builder() .put("method", method) .put("code", code) .put("message", message) .build(); byte[] data = frameEncoder.encodeCustomFrame(frame.getSlaveId(), JsonUtils.toJsonString(response), frameFormat, frame.getTransactionId()); connectionManager.sendToSocket(socket, data); } // ========== 轮询响应处理 ========== /** * 处理轮询响应(云端轮询模式) */ private void handlePollingResponse(NetSocket socket, IotModbusFrame frame, IotModbusFrameFormatEnum frameFormat) { // 1. 获取连接信息(未认证连接丢弃) ConnectionInfo info = connectionManager.getConnectionInfo(socket); if (info == null) { log.warn("[handlePollingResponse][未认证连接, 丢弃数据, remoteAddress={}]", socket.remoteAddress()); return; } // 2.1 匹配 PendingRequest PendingRequest request = pendingRequestManager.matchResponse( info.getDeviceId(), frame, frameFormat); if (request == null) { log.debug("[handlePollingResponse][未匹配到 PendingRequest, deviceId={}, FC={}]", info.getDeviceId(), frame.getFunctionCode()); return; } // 2.2 提取寄存器值 int[] rawValues = IotModbusCommonUtils.extractValues(frame); if (rawValues == null) { log.warn("[handlePollingResponse][提取寄存器值失败, deviceId={}, identifier={}]", info.getDeviceId(), request.getIdentifier()); return; } // 2.3 查找点位配置 IotModbusDeviceConfigRespDTO config = configCacheService.getConfig(info.getDeviceId()); IotModbusPointRespDTO point = IotModbusCommonUtils.findPointById(config, request.getPointId()); if (point == null) { return; } // 3.1 转换原始值为物模型属性值(点位翻译) Object convertedValue = IotModbusCommonUtils.convertToPropertyValue(rawValues, point); // 3.2 构造属性上报消息 Map params = MapUtil.of(request.getIdentifier(), convertedValue); IotDeviceMessage message = IotDeviceMessage.requestOf( IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod(), params); // 4. 发送到消息总线 messageService.sendDeviceMessage(message, info.getProductKey(), info.getDeviceName(), serverId); log.debug("[handlePollingResponse][设备={}, 属性={}, 原始值={}, 转换值={}]", info.getDeviceId(), request.getIdentifier(), rawValues, convertedValue); } }