package cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpserver.handler.downstream; import cn.hutool.core.lang.Assert; import cn.hutool.core.util.ObjUtil; 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.gateway.protocol.modbus.common.utils.IotModbusCommonUtils; 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 lombok.extern.slf4j.Slf4j; import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; /** * IoT Modbus TCP Server 下行消息处理器 *

* 负责: * 1. 处理下行消息(如属性设置 thing.service.property.set) * 2. 将属性值转换为 Modbus 写指令,通过 TCP 连接发送给设备 * * @author 芋道源码 */ @Slf4j public class IotModbusTcpServerDownstreamHandler { private final IotModbusTcpServerConnectionManager connectionManager; private final IotModbusTcpServerConfigCacheService configCacheService; private final IotModbusFrameEncoder frameEncoder; /** * TCP 事务 ID 自增器(与 PollScheduler 共享) */ private final AtomicInteger transactionIdCounter; public IotModbusTcpServerDownstreamHandler(IotModbusTcpServerConnectionManager connectionManager, IotModbusTcpServerConfigCacheService configCacheService, IotModbusFrameEncoder frameEncoder, AtomicInteger transactionIdCounter) { this.connectionManager = connectionManager; this.configCacheService = configCacheService; this.frameEncoder = frameEncoder; this.transactionIdCounter = transactionIdCounter; } /** * 处理下行消息 */ @SuppressWarnings({"unchecked", "DuplicatedCode"}) public void handle(IotDeviceMessage message) { // 1.1 检查是否是属性设置消息 if (ObjUtil.equals(IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod(), message.getMethod())) { return; } if (ObjUtil.notEqual(IotDeviceMessageMethodEnum.PROPERTY_SET.getMethod(), message.getMethod())) { log.debug("[handle][忽略非属性设置消息: {}]", message.getMethod()); return; } // 1.2 获取设备配置 IotModbusDeviceConfigRespDTO config = configCacheService.getConfig(message.getDeviceId()); if (config == null) { log.warn("[handle][设备 {} 没有 Modbus 配置]", message.getDeviceId()); return; } // 1.3 获取连接信息 ConnectionInfo connInfo = connectionManager.getConnectionInfoByDeviceId(message.getDeviceId()); if (connInfo == null) { log.warn("[handle][设备 {} 没有连接]", message.getDeviceId()); return; } // 2. 解析属性值并写入 Object params = message.getParams(); if (!(params instanceof Map)) { log.warn("[handle][params 不是 Map 类型: {}]", params); return; } Map propertyMap = (Map) params; for (Map.Entry entry : propertyMap.entrySet()) { String identifier = entry.getKey(); Object value = entry.getValue(); // 2.1 查找对应的点位配置 IotModbusPointRespDTO point = IotModbusCommonUtils.findPoint(config, identifier); if (point == null) { log.warn("[handle][设备 {} 没有点位配置: {}]", message.getDeviceId(), identifier); continue; } // 2.2 检查是否支持写操作 if (!IotModbusCommonUtils.isWritable(point.getFunctionCode())) { log.warn("[handle][点位 {} 不支持写操作, 功能码={}]", identifier, point.getFunctionCode()); continue; } // 2.3 执行写入 writeProperty(config.getDeviceId(), connInfo, point, value); } } /** * 写入属性值 */ private void writeProperty(Long deviceId, ConnectionInfo connInfo, IotModbusPointRespDTO point, Object value) { // 1.1 转换属性值为原始值 int[] rawValues = IotModbusCommonUtils.convertToRawValues(value, point); // 1.2 确定帧格式和事务 ID IotModbusFrameFormatEnum frameFormat = connInfo.getFrameFormat(); Assert.notNull(frameFormat, "连接帧格式不能为空"); Integer transactionId = frameFormat == IotModbusFrameFormatEnum.MODBUS_TCP ? (transactionIdCounter.incrementAndGet() & 0xFFFF) : null; int slaveId = connInfo.getSlaveId() != null ? connInfo.getSlaveId() : 1; // 1.3 编码写请求 byte[] data; int readFunctionCode = point.getFunctionCode(); Integer writeSingleCode = IotModbusCommonUtils.getWriteSingleFunctionCode(readFunctionCode); Integer writeMultipleCode = IotModbusCommonUtils.getWriteMultipleFunctionCode(readFunctionCode); if (rawValues.length == 1 && writeSingleCode != null) { // 单个值:使用单写功能码(FC05/FC06) data = frameEncoder.encodeWriteSingleRequest(slaveId, writeSingleCode, point.getRegisterAddress(), rawValues[0], frameFormat, transactionId); } else if (writeMultipleCode != null) { // 多个值:使用多写功能码(FC15/FC16) if (writeMultipleCode == IotModbusCommonUtils.FC_WRITE_MULTIPLE_COILS) { data = frameEncoder.encodeWriteMultipleCoilsRequest(slaveId, point.getRegisterAddress(), rawValues, frameFormat, transactionId); } else { data = frameEncoder.encodeWriteMultipleRegistersRequest(slaveId, point.getRegisterAddress(), rawValues, frameFormat, transactionId); } } else { log.warn("[writeProperty][点位 {} 不支持写操作, 功能码={}]", point.getIdentifier(), readFunctionCode); return; } // 2. 发送消息 connectionManager.sendToDevice(deviceId, data).onSuccess(v -> log.info("[writeProperty][写入成功, deviceId={}, identifier={}, value={}]", deviceId, point.getIdentifier(), value) ).onFailure(e -> log.error("[writeProperty][写入失败, deviceId={}, identifier={}, value={}]", deviceId, point.getIdentifier(), value, e) ); } }