diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/manager/AbstractIotModbusPollScheduler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/manager/AbstractIotModbusPollScheduler.java index e62f85f..a41634b 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/manager/AbstractIotModbusPollScheduler.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/manager/AbstractIotModbusPollScheduler.java @@ -12,6 +12,7 @@ import java.util.*; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentLinkedQueue; +import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertList; import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertSet; /** @@ -19,6 +20,8 @@ import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils. *

* 封装通用的定时器管理、per-device 请求队列限速逻辑。 * 子类只需实现 {@link #pollPoint(Long, Long)} 定义具体的轮询动作。 + * 如需将多个点位合并为一个轮询任务，可覆盖 {@link #buildPollTasks(IotModbusDeviceConfigRespDTO)} + * 和 {@link #pollTask(Long, String)}。 *

* * @author 芋道源码 @@ -38,9 +41,9 @@ public abstract class AbstractIotModbusPollScheduler { private static final int MAX_QUEUE_SIZE = 1000; /** - * 设备点位的定时器映射：deviceId -> (pointId -> PointTimerInfo) + * 设备轮询任务的定时器映射：deviceId -> (taskKey -> PollTimerInfo) */ - private final Map> devicePointTimers = new ConcurrentHashMap<>(); + private final Map> devicePollTimers = new ConcurrentHashMap<>(); /** * per-device 请求队列：deviceId -> 待执行请求队列 @@ -60,11 +63,29 @@ public abstract class AbstractIotModbusPollScheduler { } /** - * 点位定时器信息 + * 轮询任务信息 */ @Data @AllArgsConstructor - private static class PointTimerInfo { + protected static class PollTask { + + /** + * 任务标识 + */ + private String key; + /** + * 轮询间隔（用于判断是否需要更新定时器） + */ + private Integer pollInterval; + + } + + /** + * 轮询定时器信息 + */ + @Data + @AllArgsConstructor + private static class PollTimerInfo { /** * Vert.x 定时器 ID @@ -82,75 +103,84 @@ public abstract class AbstractIotModbusPollScheduler { /** * 更新轮询任务（增量更新） * - * 1. 【删除】点位：停止对应的轮询定时器 - * 2. 【新增】点位：创建对应的轮询定时器 - * 3. 【修改】点位：pollInterval 变化，重建对应的轮询定时器 - * 【修改】其他属性变化：不需要重建定时器（pollPoint 运行时从 configCache 取最新 point） + * 1. 【删除】任务：停止对应的轮询定时器 + * 2. 【新增】任务：创建对应的轮询定时器 + * 3. 【修改】任务：pollInterval 变化，重建对应的轮询定时器 + * 【修改】其他属性变化：不需要重建定时器（pollTask 运行时从 configCache 取最新配置） */ public void updatePolling(IotModbusDeviceConfigRespDTO config) { Long deviceId = config.getDeviceId(); - List newPoints = config.getPoints(); - Map currentTimers = devicePointTimers + List newTasks = buildPollTasks(config); + Map currentTimers = devicePollTimers .computeIfAbsent(deviceId, k -> new ConcurrentHashMap<>()); - // 1.1 计算新配置中的点位 ID 集合 - Set newPointIds = convertSet(newPoints, IotModbusPointRespDTO::getId); - // 1.2 计算删除的点位 ID 集合 - Set removedPointIds = new HashSet<>(currentTimers.keySet()); - removedPointIds.removeAll(newPointIds); - - // 2. 处理删除的点位：停止不再存在的定时器 - for (Long pointId : removedPointIds) { - PointTimerInfo timerInfo = currentTimers.remove(pointId); + // 1.1 计算新配置中的任务 Key 集合 + Set newTaskKeys = convertSet(newTasks, PollTask::getKey); + // 1.2 计算删除的任务 Key 集合 + Set removedTaskKeys = new HashSet<>(currentTimers.keySet()); + removedTaskKeys.removeAll(newTaskKeys); + + // 2. 处理删除的任务：停止不再存在的定时器 + for (String taskKey : removedTaskKeys) { + PollTimerInfo timerInfo = currentTimers.remove(taskKey); if (timerInfo != null) { vertx.cancelTimer(timerInfo.getTimerId()); - log.debug("[updatePolling][设备 {} 点位 {} 定时器已删除]", deviceId, pointId); + log.debug("[updatePolling][设备 {} 轮询任务 {} 定时器已删除]", deviceId, taskKey); } } - // 3. 处理新增和修改的点位 - if (CollUtil.isEmpty(newPoints)) { + // 3. 处理新增和修改的任务 + if (CollUtil.isEmpty(newTasks)) { return; } - for (IotModbusPointRespDTO point : newPoints) { - Long pointId = point.getId(); - Integer newPollInterval = point.getPollInterval(); - PointTimerInfo existingTimer = currentTimers.get(pointId); - // 3.1 新增点位：创建定时器 + for (PollTask task : newTasks) { + String taskKey = task.getKey(); + Integer newPollInterval = task.getPollInterval(); + PollTimerInfo existingTimer = currentTimers.get(taskKey); + // 3.1 新增任务：创建定时器 if (existingTimer == null) { - Long timerId = createPollTimer(deviceId, pointId, newPollInterval); + Long timerId = createPollTimer(deviceId, taskKey, newPollInterval); if (timerId != null) { - currentTimers.put(pointId, new PointTimerInfo(timerId, newPollInterval)); - log.debug("[updatePolling][设备 {} 点位 {} 定时器已创建, interval={}ms]", - deviceId, pointId, newPollInterval); + currentTimers.put(taskKey, new PollTimerInfo(timerId, newPollInterval)); + log.debug("[updatePolling][设备 {} 轮询任务 {} 定时器已创建, interval={}ms]", + deviceId, taskKey, newPollInterval); } } else if (!Objects.equals(existingTimer.getPollInterval(), newPollInterval)) { // 3.2 pollInterval 变化：重建定时器 vertx.cancelTimer(existingTimer.getTimerId()); - Long timerId = createPollTimer(deviceId, pointId, newPollInterval); + Long timerId = createPollTimer(deviceId, taskKey, newPollInterval); if (timerId != null) { - currentTimers.put(pointId, new PointTimerInfo(timerId, newPollInterval)); - log.debug("[updatePolling][设备 {} 点位 {} 定时器已更新, interval={}ms -> {}ms]", - deviceId, pointId, existingTimer.getPollInterval(), newPollInterval); + currentTimers.put(taskKey, new PollTimerInfo(timerId, newPollInterval)); + log.debug("[updatePolling][设备 {} 轮询任务 {} 定时器已更新, interval={}ms -> {}ms]", + deviceId, taskKey, existingTimer.getPollInterval(), newPollInterval); } else { - currentTimers.remove(pointId); + currentTimers.remove(taskKey); } } - // 3.3 其他属性变化：无需重建定时器，因为 pollPoint() 运行时从 configCache 获取最新 point，自动使用新配置 + // 3.3 其他属性变化：无需重建定时器，因为 pollTask() 运行时从 configCache 获取最新配置 } } /** + * 构建轮询任务列表 + * + * 默认每个点位一个轮询任务。TCP Client 等协议可覆盖该方法，将多个点位合并为一个批量读取任务。 + */ + protected List buildPollTasks(IotModbusDeviceConfigRespDTO config) { + return convertList(config.getPoints(), point -> new PollTask(String.valueOf(point.getId()), point.getPollInterval())); + } + + /** * 创建轮询定时器 */ - private Long createPollTimer(Long deviceId, Long pointId, Integer pollInterval) { + private Long createPollTimer(Long deviceId, String taskKey, Integer pollInterval) { if (pollInterval == null || pollInterval <= 0) { return null; } return vertx.setPeriodic(pollInterval, timerId -> { try { - submitPollRequest(deviceId, pointId); + submitPollRequest(deviceId, taskKey); } catch (Exception e) { - log.error("[createPollTimer][轮询点位失败, deviceId={}, pointId={}]", deviceId, pointId, e); + log.error("[createPollTimer][轮询任务失败, deviceId={}, taskKey={}]", deviceId, taskKey, e); } }); } @@ -160,7 +190,7 @@ public abstract class AbstractIotModbusPollScheduler { /** * 提交轮询请求到设备请求队列（保证同设备请求间隔） */ - private void submitPollRequest(Long deviceId, Long pointId) { + private void submitPollRequest(Long deviceId, String taskKey) { // 1. 【重要】将请求添加到设备的请求队列 Queue queue = deviceRequestQueues.computeIfAbsent(deviceId, k -> new ConcurrentLinkedQueue<>()); while (queue.size() >= MAX_QUEUE_SIZE) { @@ -168,7 +198,7 @@ public abstract class AbstractIotModbusPollScheduler { queue.poll(); log.warn("[submitPollRequest][设备 {} 请求队列已满({}), 丢弃最旧请求]", deviceId, MAX_QUEUE_SIZE); } - queue.offer(() -> pollPoint(deviceId, pointId)); + queue.offer(() -> pollTask(deviceId, taskKey)); // 2. 处理设备请求队列（如果没有延迟 timer 在等待） processDeviceQueue(deviceId); @@ -239,6 +269,15 @@ public abstract class AbstractIotModbusPollScheduler { // ========== 轮询执行 ========== /** + * 轮询任务 + * + * 默认将任务标识作为点位 ID，执行单点轮询。 + */ + protected void pollTask(Long deviceId, String taskKey) { + pollPoint(deviceId, Long.valueOf(taskKey)); + } + + /** * 轮询单个点位（子类实现具体的读取逻辑） * * @param deviceId 设备 ID @@ -252,25 +291,25 @@ public abstract class AbstractIotModbusPollScheduler { * 停止设备的轮询 */ public void stopPolling(Long deviceId) { - Map timers = devicePointTimers.remove(deviceId); - if (CollUtil.isEmpty(timers)) { - return; - } - for (PointTimerInfo timerInfo : timers.values()) { - vertx.cancelTimer(timerInfo.getTimerId()); + Map timers = devicePollTimers.remove(deviceId); + if (CollUtil.isNotEmpty(timers)) { + for (PollTimerInfo timerInfo : timers.values()) { + vertx.cancelTimer(timerInfo.getTimerId()); + } } // 清理请求队列 deviceRequestQueues.remove(deviceId); deviceLastRequestTime.remove(deviceId); deviceDelayTimerActive.remove(deviceId); - log.debug("[stopPolling][设备 {} 停止了 {} 个轮询定时器]", deviceId, timers.size()); + log.debug("[stopPolling][设备 {} 停止了 {} 个轮询定时器]", deviceId, + CollUtil.isEmpty(timers) ? 0 : timers.size()); } /** * 停止所有轮询 */ public void stopAll() { - for (Long deviceId : new ArrayList<>(devicePointTimers.keySet())) { + for (Long deviceId : new ArrayList<>(devicePollTimers.keySet())) { stopPolling(deviceId); } } diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/utils/IotModbusTcpClientUtils.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/utils/IotModbusTcpClientUtils.java index 1324f3a..1d52f8d 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/utils/IotModbusTcpClientUtils.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/utils/IotModbusTcpClientUtils.java @@ -39,11 +39,38 @@ public class IotModbusTcpClientUtils { public static Future read(IotModbusTcpClientConnectionManager.ModbusConnection connection, Integer slaveId, IotModbusPointRespDTO point) { + return read(connection, slaveId, point.getFunctionCode(), + point.getRegisterAddress(), point.getRegisterCount(), point.getIdentifier()); + } + + /** + * 读取 Modbus 数据 + * + * @param connection Modbus 连接 + * @param slaveId 从站地址 + * @param functionCode 功能码 + * @param registerAddress 寄存器起始地址 + * @param registerCount 寄存器数量 + * @return 原始值（int 数组） + */ + public static Future read(IotModbusTcpClientConnectionManager.ModbusConnection connection, + Integer slaveId, + Integer functionCode, + Integer registerAddress, + Integer registerCount) { + return read(connection, slaveId, functionCode, registerAddress, registerCount, null); + } + + private static Future read(IotModbusTcpClientConnectionManager.ModbusConnection connection, + Integer slaveId, + Integer functionCode, + Integer registerAddress, + Integer registerCount, + String identifier) { return connection.executeBlocking(tcpConnection -> { try { // 1. 创建请求 - ModbusRequest request = createReadRequest(point.getFunctionCode(), - point.getRegisterAddress(), point.getRegisterCount()); + ModbusRequest request = createReadRequest(functionCode, registerAddress, registerCount); request.setUnitID(slaveId); // 2. 执行事务（请求） @@ -53,10 +80,10 @@ public class IotModbusTcpClientUtils { // 3. 解析响应 ModbusResponse response = transaction.getResponse(); - return extractValues(response, point.getFunctionCode()); + return extractValues(response, functionCode); } catch (Exception e) { - throw new RuntimeException(String.format("Modbus 读取失败 [slaveId=%d, identifier=%s, address=%d]", - slaveId, point.getIdentifier(), point.getRegisterAddress()), e); + throw new RuntimeException(String.format("Modbus 读取失败 [slaveId=%d, identifier=%s, functionCode=%d, address=%d, count=%d]", + slaveId, identifier, functionCode, registerAddress, registerCount), e); } }); } diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpclient/manager/IotModbusTcpClientPollScheduler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpclient/manager/IotModbusTcpClientPollScheduler.java index 946937d..fba2f75 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpclient/manager/IotModbusTcpClientPollScheduler.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpclient/manager/IotModbusTcpClientPollScheduler.java @@ -9,8 +9,16 @@ import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.utils.IotModbu import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.utils.IotModbusTcpClientUtils; import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpclient.handler.upstream.IotModbusTcpClientUpstreamHandler; import io.vertx.core.Vertx; +import lombok.AllArgsConstructor; +import lombok.Data; import lombok.extern.slf4j.Slf4j; +import java.util.*; + +import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertList; +import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertMultiMap; +import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.filterList; + /** * IoT Modbus TCP Client 轮询调度器：管理点位的轮询定时器，调度读取任务并上报结果 * @@ -35,6 +43,47 @@ public class IotModbusTcpClientPollScheduler extends AbstractIotModbusPollSchedu // ========== 轮询执行 ========== + @Override + protected List buildPollTasks(IotModbusDeviceConfigRespDTO config) { + return convertList(buildReadSegments(config), segment -> new PollTask(segment.getKey(), segment.getPollInterval())); + } + + /** + * 轮询读取段 + */ + @Override + protected void pollTask(Long deviceId, String taskKey) { + // 1.1 从 configCache 获取最新配置 + IotModbusDeviceConfigRespDTO config = configCacheService.getConfig(deviceId); + if (config == null || CollUtil.isEmpty(config.getPoints())) { + log.warn("[pollTask][设备 {} 没有配置]", deviceId); + return; + } + // 1.2 查找读取段。配置变化后，如果当前 taskKey 已不存在，直接跳过等待下一轮 updatePolling 清理 timer + ReadSegment segment = findReadSegment(config, taskKey); + if (segment == null) { + log.debug("[pollTask][设备 {} 读取段 {} 未找到，跳过陈旧轮询任务]", deviceId, taskKey); + return; + } + + // 2.1 获取连接 + IotModbusTcpClientConnectionManager.ModbusConnection connection = connectionManager.getConnection(deviceId); + if (connection == null) { + log.warn("[pollTask][设备 {} 没有连接]", deviceId); + return; + } + // 2.2 获取 slave ID + Integer slaveId = connectionManager.getSlaveId(deviceId); + Assert.notNull(slaveId, "设备 {} 没有配置 slaveId", deviceId); + + // 3. 执行 Modbus 批量读取 + IotModbusTcpClientUtils.read(connection, slaveId, segment.getFunctionCode(), + segment.getStartAddress(), segment.getRegisterCount()) + .onSuccess(rawValues -> handleSegmentReadResult(config, segment, rawValues)) + .onFailure(e -> log.error("[pollTask][读取点位段失败, deviceId={}, segment={}]", + deviceId, segment.getKey(), e)); + } + /** * 轮询单个点位 */ @@ -70,4 +119,174 @@ public class IotModbusTcpClientPollScheduler extends AbstractIotModbusPollSchedu deviceId, point.getIdentifier(), e)); } + private void handleSegmentReadResult(IotModbusDeviceConfigRespDTO config, + ReadSegment segment, + int[] rawValues) { + for (IotModbusPointRespDTO point : segment.getPoints()) { + // 批量读取返回的是整个连续地址段，需要按点位地址切片后再复用单点上报逻辑 + int[] pointRawValues = extractPointRawValues(rawValues, segment, point); + if (pointRawValues == null) { + log.warn("[handleSegmentReadResult][读取段结果长度不足, deviceId={}, segment={}, identifier={}]", + config.getDeviceId(), segment.getKey(), point.getIdentifier()); + continue; + } + upstreamHandler.handleReadResult(config, point, pointRawValues); + } + } + + private ReadSegment findReadSegment(IotModbusDeviceConfigRespDTO config, String taskKey) { + return CollUtil.findOne(buildReadSegments(config), segment -> segment.getKey().equals(taskKey)); + } + + /** + * 构建批量读取段 + * + *

只合并功能码、轮询间隔相同，且地址连续或重叠的点位；不跨功能码、不跨轮询间隔，避免改变原有轮询语义。 + * 同时按 Modbus 协议限制控制单次读取长度，超过限制时拆成多个读取段。 + */ + static List buildReadSegments(IotModbusDeviceConfigRespDTO config) { + if (config == null) { + return Collections.emptyList(); + } + + // 1. 按功能码和轮询间隔分组：两者任一不同，都不能共用同一次 Modbus 读请求 + List validPoints = filterList(config.getPoints(), IotModbusTcpClientPollScheduler::isValidReadPoint); + if (CollUtil.isEmpty(validPoints)) { + return Collections.emptyList(); + } + Map> pointsByGroup = convertMultiMap(validPoints, + point -> new SegmentGroupKey(point.getFunctionCode(), point.getPollInterval())); + + // 2. 组内按地址排序后，合并连续或重叠区间，生成实际轮询的读取段 + List segments = new ArrayList<>(); + for (Map.Entry> entry : pointsByGroup.entrySet()) { + List points = entry.getValue(); + points.sort(Comparator.comparing(IotModbusPointRespDTO::getRegisterAddress) + .thenComparing(IotModbusPointRespDTO::getRegisterCount) + .thenComparing(IotModbusPointRespDTO::getId)); + buildReadSegments(entry.getKey(), points, segments); + } + // 3. 固定排序，保证生成的 taskKey 稳定，便于 updatePolling 做增量更新 + segments.sort(Comparator.comparing(ReadSegment::getFunctionCode) + .thenComparing(ReadSegment::getPollInterval) + .thenComparing(ReadSegment::getStartAddress)); + return segments; + } + + private static void buildReadSegments(SegmentGroupKey groupKey, + List points, + List segments) { + ReadSegment current = null; + int maxRegisterCount = getMaxRegisterCount(groupKey.getFunctionCode()); + // points 已按 registerAddress 排序，因此可以线性合并连续/重叠地址段 + for (IotModbusPointRespDTO point : points) { + int pointStartAddress = point.getRegisterAddress(); + int pointEndAddress = pointStartAddress + point.getRegisterCount(); + // 1. 当前点位无法合并时，新建一个读取段 + if (current == null || !canMerge(current, pointStartAddress, pointEndAddress, maxRegisterCount)) { + current = new ReadSegment(groupKey.getFunctionCode(), groupKey.getPollInterval(), + pointStartAddress, point.getRegisterCount(), new ArrayList<>()); + segments.add(current); + } else { + // 2. 当前点位可合并时，扩展读取段覆盖范围 + current.setRegisterCount(Math.max(current.getEndAddress(), pointEndAddress) - current.getStartAddress()); + } + // 3. 记录读取段包含的点位，读取成功后按点位逐个切片上报 + current.getPoints().add(point); + } + } + + /** + * 判断点位是否可以合并到当前读取段 + * + *

仅合并连续或重叠区间，不合并存在地址空洞的区间，避免额外读取无关寄存器。 + */ + private static boolean canMerge(ReadSegment segment, int pointStartAddress, int pointEndAddress, int maxRegisterCount) { + if (pointStartAddress > segment.getEndAddress()) { + return false; + } + int mergedRegisterCount = Math.max(segment.getEndAddress(), pointEndAddress) - segment.getStartAddress(); + return mergedRegisterCount <= maxRegisterCount; + } + + /** + * 从批量读取结果中提取单个点位的原始值 + * + *

例如读取段从地址 10 开始，点位地址为 12、数量为 2，则取 rawValues[2..4)。 + */ + static int[] extractPointRawValues(int[] rawValues, ReadSegment segment, IotModbusPointRespDTO point) { + if (rawValues == null) { + return null; + } + // 1. 计算点位在批量读取结果中的相对偏移 + int offset = point.getRegisterAddress() - segment.getStartAddress(); + int end = offset + point.getRegisterCount(); + // 2. 防御异常响应长度，避免越界影响同一读取段内其它点位 + if (offset < 0 || end > rawValues.length) { + return null; + } + // 3. 返回单个点位需要的原始寄存器值 + return Arrays.copyOfRange(rawValues, offset, end); + } + + private static boolean isValidReadPoint(IotModbusPointRespDTO point) { + return point != null + && point.getId() != null + && point.getFunctionCode() != null + && point.getRegisterAddress() != null + && point.getRegisterCount() != null + && point.getRegisterCount() > 0 + && point.getPollInterval() != null + && point.getPollInterval() > 0; + } + + @SuppressWarnings("EnhancedSwitchMigration") + private static int getMaxRegisterCount(Integer functionCode) { + switch (functionCode) { + case IotModbusCommonUtils.FC_READ_COILS: + case IotModbusCommonUtils.FC_READ_DISCRETE_INPUTS: + return 2000; + case IotModbusCommonUtils.FC_READ_HOLDING_REGISTERS: + case IotModbusCommonUtils.FC_READ_INPUT_REGISTERS: + return 125; + default: + return 0; + } + } + + /** + * 读取段分组 Key + */ + @Data + @AllArgsConstructor + static class SegmentGroupKey { + + private Integer functionCode; + private Integer pollInterval; + + } + + /** + * 一次 Modbus 批量读请求对应的连续地址段 + */ + @Data + @AllArgsConstructor + static class ReadSegment { + + private Integer functionCode; + private Integer pollInterval; + private Integer startAddress; + private Integer registerCount; + private List points; + + String getKey() { + return functionCode + ":" + pollInterval + ":" + startAddress + ":" + registerCount; + } + + int getEndAddress() { + return startAddress + registerCount; + } + + } + } diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpclient/manager/IotModbusTcpClientPollSchedulerTest.java b/yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpclient/manager/IotModbusTcpClientPollSchedulerTest.java new file mode 100644 index 0000000..214db67 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpclient/manager/IotModbusTcpClientPollSchedulerTest.java @@ -0,0 +1,147 @@ +package cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpclient.manager; + +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.modbus.tcpclient.manager.IotModbusTcpClientPollScheduler.ReadSegment; +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import static cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.utils.IotModbusCommonUtils.FC_READ_COILS; +import static cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.utils.IotModbusCommonUtils.FC_READ_HOLDING_REGISTERS; +import static cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.utils.IotModbusCommonUtils.FC_READ_INPUT_REGISTERS; +import static org.junit.jupiter.api.Assertions.*; + +/** + * {@link IotModbusTcpClientPollScheduler} 的单元测试 + * + * @author 芋道源码 + */ +public class IotModbusTcpClientPollSchedulerTest { + + @Test + public void testBuildReadSegments_mergeContinuousPoints() { + // 准备参数 + IotModbusPointRespDTO point01 = randomPoint(1L, FC_READ_HOLDING_REGISTERS, 0, 1, 1000); + IotModbusPointRespDTO point02 = randomPoint(2L, FC_READ_HOLDING_REGISTERS, 1, 2, 1000); + IotModbusPointRespDTO point03 = randomPoint(3L, FC_READ_HOLDING_REGISTERS, 4, 1, 1000); + IotModbusDeviceConfigRespDTO config = randomConfig(point01, point02, point03); + + // 调用 + List segments = IotModbusTcpClientPollScheduler.buildReadSegments(config); + + // 断言 + assertEquals(2, segments.size()); + assertEquals(0, segments.get(0).getStartAddress()); + assertEquals(3, segments.get(0).getRegisterCount()); + assertIterableEquals(Arrays.asList(point01, point02), segments.get(0).getPoints()); + assertEquals(4, segments.get(1).getStartAddress()); + assertEquals(1, segments.get(1).getRegisterCount()); + assertIterableEquals(Collections.singletonList(point03), segments.get(1).getPoints()); + } + + @Test + public void testBuildReadSegments_mergeOverlappingPoints() { + // 准备参数 + IotModbusPointRespDTO point01 = randomPoint(1L, FC_READ_HOLDING_REGISTERS, 0, 2, 1000); + IotModbusPointRespDTO point02 = randomPoint(2L, FC_READ_HOLDING_REGISTERS, 1, 1, 1000); + IotModbusDeviceConfigRespDTO config = randomConfig(point01, point02); + + // 调用 + List segments = IotModbusTcpClientPollScheduler.buildReadSegments(config); + + // 断言 + assertEquals(1, segments.size()); + assertEquals(0, segments.get(0).getStartAddress()); + assertEquals(2, segments.get(0).getRegisterCount()); + assertIterableEquals(Arrays.asList(point01, point02), segments.get(0).getPoints()); + } + + @Test + public void testBuildReadSegments_notMergeDifferentFunctionCodeOrPollInterval() { + // 准备参数 + IotModbusPointRespDTO point01 = randomPoint(1L, FC_READ_HOLDING_REGISTERS, 0, 1, 1000); + IotModbusPointRespDTO point02 = randomPoint(2L, FC_READ_INPUT_REGISTERS, 1, 1, 1000); + IotModbusPointRespDTO point03 = randomPoint(3L, FC_READ_HOLDING_REGISTERS, 1, 1, 2000); + IotModbusDeviceConfigRespDTO config = randomConfig(point01, point02, point03); + + // 调用 + List segments = IotModbusTcpClientPollScheduler.buildReadSegments(config); + + // 断言 + assertEquals(3, segments.size()); + assertEquals(FC_READ_HOLDING_REGISTERS, segments.get(0).getFunctionCode()); + assertEquals(1000, segments.get(0).getPollInterval()); + assertEquals(FC_READ_HOLDING_REGISTERS, segments.get(1).getFunctionCode()); + assertEquals(2000, segments.get(1).getPollInterval()); + assertEquals(FC_READ_INPUT_REGISTERS, segments.get(2).getFunctionCode()); + assertEquals(1000, segments.get(2).getPollInterval()); + } + + @Test + public void testBuildReadSegments_splitWhenExceedsMaxRegisterCount() { + // 准备参数 + IotModbusPointRespDTO point01 = randomPoint(1L, FC_READ_HOLDING_REGISTERS, 0, 100, 1000); + IotModbusPointRespDTO point02 = randomPoint(2L, FC_READ_HOLDING_REGISTERS, 100, 30, 1000); + IotModbusDeviceConfigRespDTO config = randomConfig(point01, point02); + + // 调用 + List segments = IotModbusTcpClientPollScheduler.buildReadSegments(config); + + // 断言 + assertEquals(2, segments.size()); + assertEquals(0, segments.get(0).getStartAddress()); + assertEquals(100, segments.get(0).getRegisterCount()); + assertEquals(100, segments.get(1).getStartAddress()); + assertEquals(30, segments.get(1).getRegisterCount()); + } + + @Test + public void testExtractPointRawValues() { + // 准备参数 + IotModbusPointRespDTO point = randomPoint(1L, FC_READ_HOLDING_REGISTERS, 12, 2, 1000); + ReadSegment segment = new ReadSegment(FC_READ_HOLDING_REGISTERS, 1000, 10, 4, Collections.singletonList(point)); + int[] rawValues = new int[]{100, 200, 300, 400}; + + // 调用 + int[] pointRawValues = IotModbusTcpClientPollScheduler.extractPointRawValues(rawValues, segment, point); + + // 断言 + assertArrayEquals(new int[]{300, 400}, pointRawValues); + } + + @Test + public void testExtractPointRawValuesForCoils() { + // 准备参数 + IotModbusPointRespDTO point = randomPoint(1L, FC_READ_COILS, 3, 2, 1000); + ReadSegment segment = new ReadSegment(FC_READ_COILS, 1000, 0, 5, Collections.singletonList(point)); + int[] rawValues = new int[]{1, 0, 1, 1, 0, 0, 0, 0}; // 线圈响应可能按字节补齐 + + // 调用 + int[] pointRawValues = IotModbusTcpClientPollScheduler.extractPointRawValues(rawValues, segment, point); + + // 断言 + assertArrayEquals(new int[]{1, 0}, pointRawValues); + } + + private static IotModbusDeviceConfigRespDTO randomConfig(IotModbusPointRespDTO... points) { + IotModbusDeviceConfigRespDTO config = new IotModbusDeviceConfigRespDTO(); + config.setDeviceId(1L); + config.setPoints(Arrays.asList(points)); + return config; + } + + private static IotModbusPointRespDTO randomPoint(Long id, Integer functionCode, Integer registerAddress, + Integer registerCount, Integer pollInterval) { + IotModbusPointRespDTO point = new IotModbusPointRespDTO(); + point.setId(id); + point.setFunctionCode(functionCode); + point.setRegisterAddress(registerAddress); + point.setRegisterCount(registerCount); + point.setPollInterval(pollInterval); + return point; + } + +}