feat：【iot】modbus-tcp 协议接入 40%：1）优化包结构；2）增加 ModbusTcpSlaveSimulatorTest 模拟从站设备

YunaiV
1 parent 593455a0
Showing 13 changed files with 222 additions and 14 deletions
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/api/device/IoTDeviceApiImpl.java
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayConfiguration.java
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpDownstreamSubscriber.java
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpUpstreamProtocol.java
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/package-info.java
yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/ModbusTcpSlaveSimulatorTest.java
@@ -22,6 +22,7 @@ import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RestController;
  
+import java.math.BigDecimal;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -74,6 +75,11 @@ public class IoTDeviceApiImpl implements IotDeviceCommonApi {
     @PostMapping(RpcConstants.RPC_API_PREFIX + "/iot/modbus/enabled-configs")
     @PermitAll
     public CommonResult<List<IotModbusDeviceConfigRespDTO>> getEnabledModbusDeviceConfigs() {
+        // TODO @芋艿：临时 mock 数据，用于测试 ModbusTcpSlaveSimulator
+        if (true) {
+            return success(buildMockModbusConfigs());
+        }
+
         // 1. 获取所有启用的 Modbus 连接配置
         List<IotDeviceModbusConfigDO> configList = modbusConfigService.getEnabledDeviceModbusConfigList();
         if (CollUtil.isEmpty(configList)) {
@@ -105,4 +111,74 @@ public class IoTDeviceApiImpl implements IotDeviceCommonApi {
         return success(result);
     }
  
+    /**
+     * 构建 Mock Modbus 配置，对接 ModbusTcpSlaveSimulator
+     *
+     * 设备：productKey=4aymZgOTOOCrDKRT, deviceName=small
+     * 物模型字段：width, height, oneThree
+     */
+    private List<IotModbusDeviceConfigRespDTO> buildMockModbusConfigs() {
+        List<IotModbusDeviceConfigRespDTO> configs = new ArrayList<>();
+
+        // 设备配置
+        IotModbusDeviceConfigRespDTO config = new IotModbusDeviceConfigRespDTO();
+        config.setDeviceId(1L);
+        config.setProductKey("4aymZgOTOOCrDKRT");
+        config.setDeviceName("small");
+        config.setIp("127.0.0.1");
+        config.setPort(5020); // 对应 ModbusTcpSlaveSimulator 的端口
+        config.setSlaveId(1);
+        config.setTimeout(3000);
+        config.setRetryInterval(5000);
+
+        // 点位配置（对应物模型字段：width, height, oneThree）
+        List<IotModbusPointRespDTO> points = new ArrayList<>();
+
+        // 点位 1：width - 读取保持寄存器地址 0（功能码 03）
+        IotModbusPointRespDTO point1 = new IotModbusPointRespDTO();
+        point1.setId(1L);
+        point1.setIdentifier("width");
+        point1.setName("宽度");
+        point1.setFunctionCode(3); // 读保持寄存器
+        point1.setRegisterAddress(0);
+        point1.setRegisterCount(1);
+        point1.setByteOrder("AB");
+        point1.setRawDataType("INT16");
+        point1.setScale(BigDecimal.ONE);
+        point1.setPollInterval(3000); // 3 秒轮询
+        points.add(point1);
+
+        // 点位 2：height - 读取保持寄存器地址 1（功能码 03）
+        IotModbusPointRespDTO point2 = new IotModbusPointRespDTO();
+        point2.setId(2L);
+        point2.setIdentifier("height");
+        point2.setName("高度");
+        point2.setFunctionCode(3); // 读保持寄存器
+        point2.setRegisterAddress(1);
+        point2.setRegisterCount(1);
+        point2.setByteOrder("AB");
+        point2.setRawDataType("INT16");
+        point2.setScale(BigDecimal.ONE);
+        point2.setPollInterval(3000); // 3 秒轮询
+        points.add(point2);
+
+        // 点位 3：oneThree - 读取保持寄存器地址 2（功能码 03）
+        IotModbusPointRespDTO point3 = new IotModbusPointRespDTO();
+        point3.setId(3L);
+        point3.setIdentifier("oneThree");
+        point3.setName("一三");
+        point3.setFunctionCode(3); // 读保持寄存器
+        point3.setRegisterAddress(2);
+        point3.setRegisterCount(1);
+        point3.setByteOrder("AB");
+        point3.setRawDataType("INT16");
+        point3.setScale(BigDecimal.ONE);
+        point3.setPollInterval(3000); // 3 秒轮询
+        points.add(point3);
+
+        config.setPoints(points);
+        configs.add(config);
+        return configs;
+    }
+
 }
 \ No newline at end of file
@@ -7,6 +7,13 @@ import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.IotEmqxUpstreamProtocol
 import cn.iocoder.yudao.module.iot.gateway.protocol.http.IotHttpDownstreamSubscriber;
 import cn.iocoder.yudao.module.iot.gateway.protocol.http.IotHttpUpstreamProtocol;
 import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.*;
+import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.client.IotModbusTcpClient;
+import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.codec.IotModbusDataConverter;
+import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.manager.IotModbusTcpConfigCacheService;
+import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.manager.IotModbusTcpConnectionManager;
+import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.manager.IotModbusTcpPollScheduler;
+import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.router.IotModbusTcpDownstreamHandler;
+import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.router.IotModbusTcpUpstreamHandler;
 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttDownstreamSubscriber;
 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttUpstreamProtocol;
 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.manager.IotMqttConnectionManager;
@@ -4,6 +4,7 @@ import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
 import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber;
 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
 import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
+import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.router.IotModbusTcpDownstreamHandler;
 import jakarta.annotation.PostConstruct;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -2,6 +2,10 @@ package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp;
  
 import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusDeviceConfigRespDTO;
 import cn.iocoder.yudao.module.iot.gateway.config.IotGatewayProperties;
+import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.manager.IotModbusTcpConfigCacheService;
+import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.manager.IotModbusTcpConnectionManager;
+import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.manager.IotModbusTcpPollScheduler;
+import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.router.IotModbusTcpUpstreamHandler;
 import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
 import io.vertx.core.Vertx;
 import jakarta.annotation.PostConstruct;
-package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp;
+package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.client;
  
 import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusPointRespDTO;
 import cn.iocoder.yudao.module.iot.core.enums.IotModbusFunctionCodeEnum;
+import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.manager.IotModbusTcpConnectionManager;
 import com.ghgande.j2mod.modbus.io.ModbusTCPTransaction;
 import com.ghgande.j2mod.modbus.msg.*;
 import com.ghgande.j2mod.modbus.procimg.InputRegister;
@@ -12,7 +13,9 @@ import io.vertx.core.Future;
 import lombok.extern.slf4j.Slf4j;
  
 /**
- * IoT Modbus TCP 客户端，负责：
+ * IoT Modbus TCP 客户端
+ * <p>
+ * 封装 Modbus 协议读写操作：
  * 1. 封装 Modbus 读/写操作
  * 2. 根据功能码，执行对应的 Modbus 请求
  *
-package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp;
+package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.codec;
  
 import cn.hutool.core.util.ArrayUtil;
 import cn.hutool.core.util.ObjectUtil;
@@ -14,7 +14,9 @@ import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
  
 /**
- * IoT Modbus 数据转换器，负责：
+ * IoT Modbus 数据转换器
+ * <p>
+ * 负责 Modbus 原始寄存器值与物模型属性值的相互转换：
  * 1. 将 Modbus 原始寄存器值转换为物模型属性值
  * 2. 将物模型属性值转换为 Modbus 原始寄存器值
  *
-package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp;
+package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.manager;
  
 import cn.iocoder.yudao.framework.common.pojo.CommonResult;
 import cn.iocoder.yudao.module.iot.core.biz.IotDeviceCommonApi;
-package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp;
+package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.manager;
  
 import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusDeviceConfigRespDTO;
 import com.ghgande.j2mod.modbus.net.TCPMasterConnection;
@@ -16,7 +16,9 @@ import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
  
 /**
- * IoT Modbus TCP 连接管理器，负责：
+ * IoT Modbus TCP 连接管理器
+ * <p>
+ * 统一管理 Modbus TCP 连接：
  * 1. 管理 TCP 连接（相同 ip:port 共用连接）
  * 2. 分布式锁管理（连接级别），避免多节点重复创建连接
  * 3. 连接重试和故障恢复
-package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp;
+package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.manager;
  
 import cn.hutool.core.collection.CollUtil;
 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.gateway.protocol.modbustcp.router.IotModbusTcpUpstreamHandler;
+import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.client.IotModbusTcpClient;
 import io.vertx.core.Vertx;
 import lombok.AllArgsConstructor;
 import lombok.Data;
+/**
+ * Modbus TCP 协议实现包
+ * <p>
+ * 提供基于 j2mod 的 Modbus TCP 主站（Master）功能，支持：
+ * 1. 定时轮询 Modbus 从站设备数据
+ * 2. 下发属性设置命令到从站设备
+ * 3. 数据格式转换（寄存器值 ↔ 物模型属性值）
+ */
+package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp;
 \ No newline at end of file
-package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp;
+package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.router;
  
 import cn.hutool.core.collection.CollUtil;
 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.IotModbusFunctionCodeEnum;
 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
+import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.codec.IotModbusDataConverter;
+import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.client.IotModbusTcpClient;
+import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.manager.IotModbusTcpConfigCacheService;
+import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.manager.IotModbusTcpConnectionManager;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
  
@@ -34,6 +38,7 @@ public class IotModbusTcpDownstreamHandler {
     @SuppressWarnings("unchecked")
     public void handle(IotDeviceMessage message) {
         // 1.1 检查是否是属性设置消息
+        // TODO @AI：要使用枚举
         if (!"thing.service.property.set".equals(message.getMethod())) {
             log.debug("[handle][忽略非属性设置消息: {}]", message.getMethod());
             return;
-package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp;
+package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.router;
  
+import cn.hutool.core.map.MapUtil;
 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.mq.message.IotDeviceMessage;
+import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.codec.IotModbusDataConverter;
 import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
 import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
  
-import java.util.HashMap;
 import java.util.Map;
  
 /**
@@ -46,9 +48,8 @@ public class IotModbusTcpUpstreamHandler {
             log.debug("[handleReadResult][设备={}, 属性={}, 原始值={}, 转换值={}]",
                     config.getDeviceId(), point.getIdentifier(), rawValue, convertedValue);
             // 1.2 构造属性上报消息
-            Map<String, Object> params = new HashMap<>();
-            params.put(point.getIdentifier(), convertedValue);
-            IotDeviceMessage message = IotDeviceMessage.requestOf("thing.event.property.post", params);
+            Map<String, Object> params = MapUtil.of(point.getIdentifier(), convertedValue);
+            IotDeviceMessage message = IotDeviceMessage.requestOf(IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod(), params);
  
             // 2. 发送到消息总线
             messageService.sendDeviceMessage(message, config.getProductKey(),
+package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp;
+
+import com.ghgande.j2mod.modbus.procimg.*;
+import com.ghgande.j2mod.modbus.slave.ModbusSlave;
+import com.ghgande.j2mod.modbus.slave.ModbusSlaveFactory;
+
+/**
+ * Modbus TCP 从站模拟器
+ *
+ * 用于测试 Modbus TCP 网关的连接和数据读写功能
+ *
+ * @author 芋道源码
+ */
+public class ModbusTcpSlaveSimulatorTest {
+
+    private static final int PORT = 5020;
+    private static final int SLAVE_ID = 1;
+
+    @SuppressWarnings({"InfiniteLoopStatement", "BusyWait", "CommentedOutCode"})
+    public static void main(String[] args) throws Exception {
+        // 1. 创建进程映像（Process Image），存储寄存器数据
+        SimpleProcessImage spi = new SimpleProcessImage(SLAVE_ID);
+
+        // 2.1 初始化线圈（Coil，功能码 01/05）- 离散输出，可读写
+        // 地址 0-9，共 10 个线圈
+        for (int i = 0; i < 10; i++) {
+            spi.addDigitalOut(new SimpleDigitalOut(i % 2 == 0)); // 交替 true/false
+        }
+
+        // 2.2 初始化离散输入（Discrete Input，功能码 02）- 只读
+        // 地址 0-9，共 10 个离散输入
+        for (int i = 0; i < 10; i++) {
+            spi.addDigitalIn(new SimpleDigitalIn(i % 3 == 0)); // 每 3 个一个 true
+        }
+
+        // 2.3 初始化保持寄存器（Holding Register，功能码 03/06/16）- 可读写
+        // 地址 0-19，共 20 个寄存器
+        for (int i = 0; i < 20; i++) {
+            spi.addRegister(new SimpleRegister(i * 100)); // 值为 0, 100, 200, ...
+        }
+
+        // 2.4 初始化输入寄存器（Input Register，功能码 04）- 只读
+        // 地址 0-19，共 20 个寄存器
+        SimpleInputRegister[] inputRegisters = new SimpleInputRegister[20];
+        for (int i = 0; i < 20; i++) {
+            inputRegisters[i] = new SimpleInputRegister(i * 10 + 1); // 值为 1, 11, 21, ...
+            spi.addInputRegister(inputRegisters[i]);
+        }
+
+        // 3.1 创建 Modbus TCP 从站
+        ModbusSlave slave = ModbusSlaveFactory.createTCPSlave(PORT, 5);
+        slave.addProcessImage(SLAVE_ID, spi);
+        // 3.2 启动从站
+        slave.open();
+
+        System.out.println("===================================================");
+        System.out.println("Modbus TCP 从站模拟器已启动");
+        System.out.println("端口: " + PORT);
+        System.out.println("从站地址 (Slave ID): " + SLAVE_ID);
+        System.out.println("===================================================");
+        System.out.println("可用寄存器:");
+        System.out.println("  - 线圈 (Coil, 功能码 01/05): 地址 0-9");
+        System.out.println("  - 离散输入 (Discrete Input, 功能码 02): 地址 0-9");
+        System.out.println("  - 保持寄存器 (Holding Register, 功能码 03/06/16): 地址 0-19");
+        System.out.println("  - 输入寄存器 (Input Register, 功能码 04): 地址 0-19");
+        System.out.println("===================================================");
+        System.out.println("按 Ctrl+C 停止模拟器");
+
+        // 4. 添加关闭钩子
+        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+            System.out.println("\n正在关闭模拟器...");
+            slave.close();
+            System.out.println("模拟器已关闭");
+        }));
+
+        // 5. 保持运行，定时更新输入寄存器模拟数据变化
+        int counter = 0;
+        while (true) {
+            Thread.sleep(5000); // 每 5 秒更新一次
+            counter++;
+
+            // 更新输入寄存器的值，模拟传感器数据变化
+            for (int i = 0; i < 20; i++) {
+                int newValue = (i * 10 + 1) + counter;
+                inputRegisters[i].setValue(newValue);
+            }
+
+            // 更新保持寄存器的第一个值
+            spi.getRegister(0).setValue(counter * 100);
+//            System.out.println("[" + java.time.LocalTime.now() + "] 数据已更新, counter=" + counter
+//                    + ", 保持寄存器[0]=" + (counter * 100)
+//                    + ", 输入寄存器[0]=" + (1 + counter));
+        }
+    }
+
+}