From c2eefadb8239df552ef43d7ba55d5735bb603ed1 Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期日, 30 八月 2026 14:50:28 +0800
Subject: [PATCH] 实现测试暂停继续接收区间
---
ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TTestServiceImpl.java | 163 +++++++++++++++++++++++
ruoyi-admin/src/main/java/com/ruoyi/web/controller/system/TTestController.java | 30 ++++
ruoyi-admin/src/test/java/com/ruoyi/system/service/impl/TTestServiceImplTest.java | 93 +++++++++++++
ruoyi-admin/src/test/java/com/ruoyi/web/controller/system/TTestControllerContractTest.java | 9 +
ruoyi-system/src/main/resources/mapper/system/TTestMapper.xml | 48 ++++++
ruoyi-system/src/main/java/com/ruoyi/system/service/ITTestService.java | 9 +
ruoyi-system/src/main/java/com/ruoyi/system/mapper/TTestMapper.java | 11 +
ruoyi-admin/src/main/resources/db/migration/V15__pause_resume_reception_actions.sql | 27 +++
8 files changed, 390 insertions(+), 0 deletions(-)
diff --git a/ruoyi-admin/src/main/java/com/ruoyi/web/controller/system/TTestController.java b/ruoyi-admin/src/main/java/com/ruoyi/web/controller/system/TTestController.java
index c1cd2e5..db97598 100644
--- a/ruoyi-admin/src/main/java/com/ruoyi/web/controller/system/TTestController.java
+++ b/ruoyi-admin/src/main/java/com/ruoyi/web/controller/system/TTestController.java
@@ -20,6 +20,7 @@
import com.ruoyi.common.enums.BusinessType;
import com.ruoyi.system.domain.TTest;
import com.ruoyi.system.domain.TTestResult;
+import com.ruoyi.system.domain.TTestReceptionInterval;
import com.ruoyi.system.service.ITTestService;
/** 测试管理 Controller。 */
@@ -95,6 +96,35 @@
return success(testService.selectTestById(testId));
}
+ /** 独立暂停接收动作；只关闭系统接收区间，不向设备发送控制命令。 */
+ @PreAuthorize("@ss.hasPermi('system:test:pause')")
+ @Log(title = "测试数据接收", businessType = BusinessType.UPDATE)
+ @PostMapping("/{testId}/reception/pause")
+ public AjaxResult pauseReception(@PathVariable Long testId)
+ {
+ testService.pauseReception(testId, getUserId(), getUsername());
+ return success(testService.selectTestById(testId));
+ }
+
+ /** 独立继续接收动作；从新的成功观测边界打开区间，不向设备发送控制命令。 */
+ @PreAuthorize("@ss.hasPermi('system:test:resume')")
+ @Log(title = "测试数据接收", businessType = BusinessType.UPDATE)
+ @PostMapping("/{testId}/reception/resume")
+ public AjaxResult resumeReception(@PathVariable Long testId)
+ {
+ testService.resumeReception(testId, getUserId(), getUsername());
+ return success(testService.selectTestById(testId));
+ }
+
+ /** 返回测试各来源的接收区间、动作时间和实际采样边界。 */
+ @PreAuthorize("@ss.hasPermi('system:test:query')")
+ @GetMapping("/{testId}/reception/intervals")
+ public AjaxResult receptionIntervals(@PathVariable Long testId)
+ {
+ List<TTestReceptionInterval> intervals = testService.selectReceptionIntervals(testId);
+ return success(intervals);
+ }
+
@PreAuthorize("@ss.hasPermi('system:test:remove')")
@Log(title = "测试管理", businessType = BusinessType.DELETE)
@DeleteMapping("/{testIds}")
diff --git a/ruoyi-admin/src/main/resources/db/migration/V15__pause_resume_reception_actions.sql b/ruoyi-admin/src/main/resources/db/migration/V15__pause_resume_reception_actions.sql
new file mode 100644
index 0000000..7379b1a
--- /dev/null
+++ b/ruoyi-admin/src/main/resources/db/migration/V15__pause_resume_reception_actions.sql
@@ -0,0 +1,27 @@
+-- Ticket10：暂停/继续接收权限与区间查询索引；不改变已应用迁移。
+ALTER TABLE t_test_reception_interval
+ ADD KEY idx_reception_interval_status (test_id, interval_status, interval_id);
+
+SET @pause_menu_id := (SELECT COALESCE(MAX(menu_id), 0) + 1 FROM sys_menu);
+INSERT INTO sys_menu
+ (menu_id, menu_name, parent_id, order_num, path, component, query, route_name,
+ is_frame, is_cache, menu_type, visible, status, perms, icon,
+ create_by, create_time, update_by, update_time, remark)
+SELECT @pause_menu_id, '暂停接收', menu_id, 7, '', NULL, NULL, '', 1, 0, 'F', '0', '0',
+ 'system:test:pause', '#', 'admin', NOW(), '', NULL, '测试暂停接收权限'
+FROM sys_menu
+WHERE perms = 'system:test:list'
+ AND NOT EXISTS (SELECT 1 FROM sys_menu existing WHERE existing.perms = 'system:test:pause')
+LIMIT 1;
+
+SET @resume_menu_id := (SELECT COALESCE(MAX(menu_id), 0) + 1 FROM sys_menu);
+INSERT INTO sys_menu
+ (menu_id, menu_name, parent_id, order_num, path, component, query, route_name,
+ is_frame, is_cache, menu_type, visible, status, perms, icon,
+ create_by, create_time, update_by, update_time, remark)
+SELECT @resume_menu_id, '继续接收', menu_id, 8, '', NULL, NULL, '', 1, 0, 'F', '0', '0',
+ 'system:test:resume', '#', 'admin', NOW(), '', NULL, '测试继续接收权限'
+FROM sys_menu
+WHERE perms = 'system:test:list'
+ AND NOT EXISTS (SELECT 1 FROM sys_menu existing WHERE existing.perms = 'system:test:resume')
+LIMIT 1;
diff --git a/ruoyi-admin/src/test/java/com/ruoyi/system/service/impl/TTestServiceImplTest.java b/ruoyi-admin/src/test/java/com/ruoyi/system/service/impl/TTestServiceImplTest.java
index 04ec96a..c54c680 100644
--- a/ruoyi-admin/src/test/java/com/ruoyi/system/service/impl/TTestServiceImplTest.java
+++ b/ruoyi-admin/src/test/java/com/ruoyi/system/service/impl/TTestServiceImplTest.java
@@ -15,12 +15,16 @@
import java.math.BigDecimal;
import java.util.Collections;
import java.util.LinkedHashMap;
+import java.util.List;
import java.time.Instant;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
+import com.ruoyi.device.bts.BtsDevelopmentSimulator;
+import com.ruoyi.device.simulator.DevelopmentCollectorSimulation.CollectorRealtimeSnapshot;
+import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorRealtimeMeasurement;
import org.springframework.dao.DuplicateKeyException;
import com.ruoyi.common.exception.ServiceException;
import com.ruoyi.system.domain.TTest;
@@ -31,6 +35,7 @@
import com.ruoyi.system.domain.TTestPlanStep;
import com.ruoyi.system.domain.TTestWarningThreshold;
import com.ruoyi.system.domain.TDetectionChannel;
+import com.ruoyi.system.domain.TTestReceptionInterval;
import com.ruoyi.system.mapper.TTestMapper;
import com.ruoyi.system.service.ISharedFieldSnapshotService;
@@ -43,6 +48,9 @@
@Mock
private ISharedFieldSnapshotService sharedFieldSnapshotService;
+
+ @Mock
+ private BtsDevelopmentSimulator collectorSimulator;
@InjectMocks
private TTestServiceImpl testService;
@@ -488,6 +496,91 @@
verify(testMapper, never()).selectTestByIdForUpdate(anyLong());
}
+ @Test
+ void pauseReceptionShouldCloseAllOpenIntervalsAndAppendAudit()
+ {
+ TTest current = validTest();
+ current.setTestId(10L);
+ current.setReceptionStatus("RECEIVING");
+ when(testMapper.selectTestByIdForUpdate(10L)).thenReturn(current);
+ TTestReceptionInterval cabinet = receptionInterval(1L, "CABINET_CHANNEL", 100L, null);
+ TTestReceptionInterval collector = receptionInterval(2L, "COLLECTOR_CHANNEL", 8L, "S-01");
+ when(testMapper.selectOpenReceptionIntervalsForUpdate(10L)).thenReturn(List.of(cabinet, collector));
+ TDetectionChannel channel = new TDetectionChannel();
+ channel.setChannelId(100L); channel.setOnlineStatus("ONLINE"); channel.setLastSeenAtUtc(Instant.parse("2026-08-29T12:00:00Z"));
+ when(testMapper.selectReceptionChannelForUpdate(100L)).thenReturn(channel);
+ when(testMapper.collectorOnline(8L)).thenReturn(true);
+ mockCollectorBoundary();
+ when(testMapper.closeReceptionInterval(anyLong(), any(), any())).thenReturn(1);
+ when(testMapper.updateReceptionPaused(eq(10L), eq("operator"), any())).thenReturn(1);
+
+ assertEquals(1, testService.pauseReception(10L, 7L, "operator"));
+ verify(testMapper, times(2)).closeReceptionInterval(anyLong(), any(), any());
+ verify(testMapper).updateReceptionPaused(eq(10L), eq("operator"), any());
+ verify(testMapper).insertTestAudit(eq(10L), eq("RECEPTION_PAUSED"), eq(7L), eq("operator"), isNull(), any());
+ }
+
+ @Test
+ void pauseReceptionShouldFailBeforeWritesWhenBoundaryUnavailable()
+ {
+ TTest current = validTest();
+ current.setTestId(10L); current.setReceptionStatus("RECEIVING");
+ when(testMapper.selectTestByIdForUpdate(10L)).thenReturn(current);
+ TTestReceptionInterval interval = receptionInterval(1L, "CABINET_CHANNEL", 100L, null);
+ when(testMapper.selectOpenReceptionIntervalsForUpdate(10L)).thenReturn(List.of(interval));
+ when(testMapper.selectReceptionChannelForUpdate(100L)).thenReturn(null);
+
+ ServiceException exception = assertThrows(ServiceException.class,
+ () -> testService.pauseReception(10L, 7L, "operator"));
+ assertEquals("检测柜通道 100 当前无法确认暂停接收边界", exception.getMessage());
+ verify(testMapper, never()).closeReceptionInterval(anyLong(), any(), any());
+ verify(testMapper, never()).updateReceptionPaused(anyLong(), any(), any());
+ }
+
+ @Test
+ void resumeReceptionShouldOpenNewIntervalsAndAppendAudit()
+ {
+ TTest current = validTest();
+ current.setTestId(10L); current.setReceptionStatus("PAUSED");
+ current.getBatteries().get(0).getBinding().setBindingId(30L);
+ when(testMapper.selectTestByIdForUpdate(10L)).thenReturn(current);
+ TDetectionChannel channel = new TDetectionChannel();
+ channel.setChannelId(100L); channel.setOnlineStatus("ONLINE"); channel.setLastSeenAtUtc(Instant.parse("2026-08-29T12:00:00Z"));
+ when(testMapper.selectReceptionChannelForUpdate(100L)).thenReturn(channel);
+ when(testMapper.countCollectorEnabled(8L)).thenReturn(1);
+ when(testMapper.collectorOnline(8L)).thenReturn(true);
+ when(testMapper.selectCollectorChannelCount(8L)).thenReturn(4);
+ mockCollectorBoundary();
+ when(testMapper.updateReceptionResumed(eq(10L), eq("operator"), any())).thenReturn(1);
+
+ assertEquals(1, testService.resumeReception(10L, 7L, "operator"));
+ verify(testMapper, times(4)).insertReceptionInterval(any());
+ verify(testMapper).updateReceptionResumed(eq(10L), eq("operator"), any());
+ verify(testMapper).insertTestAudit(eq(10L), eq("RECEPTION_RESUMED"), eq(7L), eq("operator"), isNull(), any());
+ }
+
+ private TTestReceptionInterval receptionInterval(Long id, String sourceType, Long sourceId, String channelCode)
+ {
+ TTestReceptionInterval interval = new TTestReceptionInterval();
+ interval.setIntervalId(id); interval.setTestId(10L); interval.setSourceType(sourceType);
+ interval.setSourceId(sourceId); interval.setSourceChannelCode(channelCode); interval.setIntervalStatus("OPEN");
+ return interval;
+ }
+
+ private void mockCollectorBoundary()
+ {
+ Instant deviceTime = Instant.parse("2026-08-29T12:00:00Z");
+ CollectorRealtimeMeasurement measurement = new CollectorRealtimeMeasurement("dev-sim", "collector-8", "S-01",
+ 1L, deviceTime, "2026-08-29 20:00:00", deviceTime, "PRIMARY_TEMPERATURE", new BigDecimal("25"), "℃");
+ CollectorRealtimeMeasurement positive = new CollectorRealtimeMeasurement("dev-sim", "collector-8", "S-02",
+ 1L, deviceTime, "2026-08-29 20:00:00", deviceTime, "PRIMARY_POSITIVE_POTENTIAL", new BigDecimal("1"), "V");
+ CollectorRealtimeMeasurement negative = new CollectorRealtimeMeasurement("dev-sim", "collector-8", "S-03",
+ 1L, deviceTime, "2026-08-29 20:00:00", deviceTime, "PRIMARY_NEGATIVE_POTENTIAL", new BigDecimal("1"), "V");
+ when(collectorSimulator.collectorRealtime(8)).thenReturn(
+ new CollectorRealtimeSnapshot("collector-8", true, 1000,
+ List.of(measurement, positive, negative)));
+ }
+
private void mockEnabledOptions()
{
mockEnabledPeopleAndType();
diff --git a/ruoyi-admin/src/test/java/com/ruoyi/web/controller/system/TTestControllerContractTest.java b/ruoyi-admin/src/test/java/com/ruoyi/web/controller/system/TTestControllerContractTest.java
index 8db451e..0c6a21b 100644
--- a/ruoyi-admin/src/test/java/com/ruoyi/web/controller/system/TTestControllerContractTest.java
+++ b/ruoyi-admin/src/test/java/com/ruoyi/web/controller/system/TTestControllerContractTest.java
@@ -4,6 +4,7 @@
import java.lang.reflect.Method;
import org.junit.jupiter.api.Test;
import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.security.access.prepost.PreAuthorize;
class TTestControllerContractTest
@@ -18,6 +19,14 @@
assertPermission("startReception", "@ss.hasPermi('system:test:start')", Long.class);
Method startReception = TTestController.class.getMethod("startReception", Long.class);
assertEquals("/{testId}/reception/start", startReception.getAnnotation(PostMapping.class).value()[0]);
+ assertPermission("pauseReception", "@ss.hasPermi('system:test:pause')", Long.class);
+ assertEquals("/{testId}/reception/pause", TTestController.class.getMethod("pauseReception", Long.class)
+ .getAnnotation(PostMapping.class).value()[0]);
+ assertPermission("resumeReception", "@ss.hasPermi('system:test:resume')", Long.class);
+ assertEquals("/{testId}/reception/resume", TTestController.class.getMethod("resumeReception", Long.class)
+ .getAnnotation(PostMapping.class).value()[0]);
+ assertEquals("/{testId}/reception/intervals", TTestController.class.getMethod("receptionIntervals", Long.class)
+ .getAnnotation(GetMapping.class).value()[0]);
}
private void assertPermission(String methodName, String permission, Class<?>... parameterTypes) throws Exception
diff --git a/ruoyi-system/src/main/java/com/ruoyi/system/mapper/TTestMapper.java b/ruoyi-system/src/main/java/com/ruoyi/system/mapper/TTestMapper.java
index 47cd067..1f18933 100644
--- a/ruoyi-system/src/main/java/com/ruoyi/system/mapper/TTestMapper.java
+++ b/ruoyi-system/src/main/java/com/ruoyi/system/mapper/TTestMapper.java
@@ -78,4 +78,15 @@
int updateReceptionStarted(@Param("testId") Long testId, @Param("startedAtUtc") Date startedAtUtc,
@Param("updateBy") String updateBy);
int insertReceptionInterval(TTestReceptionInterval interval);
+ List<TTestReceptionInterval> selectOpenReceptionIntervalsForUpdate(Long testId);
+ List<TTestReceptionInterval> selectReceptionIntervals(Long testId);
+ TTestReceptionInterval selectLatestReceptionIntervalForUpdate(@Param("testId") Long testId,
+ @Param("sourceType") String sourceType, @Param("sourceId") Long sourceId,
+ @Param("sourceChannelCode") String sourceChannelCode);
+ int closeReceptionInterval(@Param("intervalId") Long intervalId, @Param("endBoundaryAtUtc") Date endBoundaryAtUtc,
+ @Param("boundaryNote") String boundaryNote);
+ int updateReceptionPaused(@Param("testId") Long testId, @Param("updateBy") String updateBy,
+ @Param("updatedAtUtc") Date updatedAtUtc);
+ int updateReceptionResumed(@Param("testId") Long testId, @Param("updateBy") String updateBy,
+ @Param("updatedAtUtc") Date updatedAtUtc);
}
diff --git a/ruoyi-system/src/main/java/com/ruoyi/system/service/ITTestService.java b/ruoyi-system/src/main/java/com/ruoyi/system/service/ITTestService.java
index e09428a..bb3082a 100644
--- a/ruoyi-system/src/main/java/com/ruoyi/system/service/ITTestService.java
+++ b/ruoyi-system/src/main/java/com/ruoyi/system/service/ITTestService.java
@@ -29,6 +29,15 @@
/** 原子开始接收：校验全部来源并从实际观测边界打开区间。 */
int startReception(Long testId, Long operatorId, String updateBy);
+ /** 原子暂停接收：关闭全部开放区间并保留设备实时观察。 */
+ int pauseReception(Long testId, Long operatorId, String updateBy);
+
+ /** 原子继续接收：从新的成功观测边界打开全部来源区间。 */
+ int resumeReception(Long testId, Long operatorId, String updateBy);
+
+ /** 查询测试各数据来源的接收区间和实际边界。 */
+ List<com.ruoyi.system.domain.TTestReceptionInterval> selectReceptionIntervals(Long testId);
+
/** 仅删除未开始接收的测试并追加审计记录。 */
int deleteTestByIds(Long[] testIds, Long operatorId, String updateBy);
diff --git a/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TTestServiceImpl.java b/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TTestServiceImpl.java
index e2fedad..a60fe2f 100644
--- a/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TTestServiceImpl.java
+++ b/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TTestServiceImpl.java
@@ -27,6 +27,8 @@
import com.ruoyi.system.domain.TDetectionChannel;
import com.ruoyi.system.domain.TTestReceptionInterval;
import com.ruoyi.system.domain.vo.TestConfigurationOptionsVo;
+import com.ruoyi.device.bts.BtsDevelopmentSimulator;
+import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorRealtimeMeasurement;
import com.ruoyi.system.mapper.TTestMapper;
import com.ruoyi.system.service.ITTestService;
import com.ruoyi.system.service.ISharedFieldSnapshotService;
@@ -49,6 +51,9 @@
@Autowired
private ISharedFieldSnapshotService sharedFieldSnapshotService;
+
+ @Autowired(required = false)
+ private BtsDevelopmentSimulator collectorSimulator;
@Override
public TTest selectTestById(Long testId)
@@ -229,6 +234,164 @@
}
}
+ @Override
+ @Transactional(rollbackFor = Exception.class)
+ public int pauseReception(Long testId, Long operatorId, String updateBy)
+ {
+ if (testId == null) throw new ServiceException("测试ID不能为空");
+ if (operatorId == null) throw new ServiceException("操作人不能为空");
+ TTest current = testMapper.selectTestByIdForUpdate(testId);
+ if (current == null) throw new ServiceException("测试不存在");
+ if (!"RECEIVING".equals(current.getReceptionStatus()))
+ throw new ServiceException("当前测试不是接收中状态");
+
+ List<TTestReceptionInterval> openIntervals = mutable(testMapper.selectOpenReceptionIntervalsForUpdate(testId));
+ if (openIntervals.isEmpty()) throw new ServiceException("当前测试没有开放的接收区间");
+ Date actionAt = new Date();
+ Map<Long, Date> boundaries = new HashMap<>();
+ // 先确认所有来源都能取得边界，再写入任何一条关闭记录，保证整项动作原子失败。
+ for (TTestReceptionInterval open : openIntervals)
+ {
+ Date boundary = confirmBoundary(open, actionAt);
+ boundaries.put(open.getIntervalId(), boundary);
+ open.setEndBoundaryAtUtc(boundary);
+ open.setIntervalStatus("CLOSED");
+ }
+ for (TTestReceptionInterval open : openIntervals)
+ {
+ if (testMapper.closeReceptionInterval(open.getIntervalId(), boundaries.get(open.getIntervalId()),
+ "暂停接收：采用来源最近成功观测边界") != 1)
+ throw new ServiceException("接收区间已变更，请刷新后重试");
+ }
+ if (testMapper.updateReceptionPaused(testId, updateBy, actionAt) != 1)
+ throw new ServiceException("测试状态已变更，请刷新后重试");
+ testMapper.insertTestAudit(testId, "RECEPTION_PAUSED", operatorId, updateBy, null,
+ JSON.toJSONString(Map.of("receptionStatus", "PAUSED", "intervalCount", openIntervals.size(),
+ "intervals", openIntervals)));
+ return 1;
+ }
+
+ @Override
+ @Transactional(rollbackFor = Exception.class)
+ public int resumeReception(Long testId, Long operatorId, String updateBy)
+ {
+ if (testId == null) throw new ServiceException("测试ID不能为空");
+ if (operatorId == null) throw new ServiceException("操作人不能为空");
+ TTest current = testMapper.selectTestByIdForUpdate(testId);
+ if (current == null) throw new ServiceException("测试不存在");
+ if (!"PAUSED".equals(current.getReceptionStatus()))
+ throw new ServiceException("当前测试不是暂停接收状态");
+ loadConfiguration(current);
+ if (current.getBatteries().isEmpty()) throw new ServiceException("测试未配置电池号");
+
+ Date actionAt = new Date();
+ List<TTestReceptionInterval> intervals = new ArrayList<>();
+ Set<Long> channelIds = new HashSet<>();
+ Set<String> collectorChannels = new HashSet<>();
+ for (TTestBattery battery : current.getBatteries())
+ {
+ TTestChannelBinding binding = battery.getBinding();
+ if (binding == null || binding.getChannelId() == null)
+ throw new ServiceException("电池号 " + battery.getBatteryNo() + " 未绑定检测柜通道");
+ if (!channelIds.add(binding.getChannelId())) throw new ServiceException("检测柜通道重复绑定，无法继续接收");
+ TDetectionChannel channel = testMapper.selectReceptionChannelForUpdate(binding.getChannelId());
+ if (channel == null || !isEnabled(channel.getStatus()) || !isOnline(channel.getOnlineStatus())
+ || channel.getLastSeenAtUtc() == null)
+ throw new ServiceException("检测柜通道 " + binding.getChannelId() + " 当前无法确认继续接收边界");
+ if (isEndedWorkStatus(channel.getWorkStatus()))
+ throw new ServiceException("检测柜通道 " + binding.getChannelId() + " 已结束，不能继续接收");
+ if (battery.getCollectorId() == null || testMapper.countCollectorEnabled(battery.getCollectorId()) == 0
+ || !testMapper.collectorOnline(battery.getCollectorId()))
+ throw new ServiceException("采集器 " + battery.getCollectorId() + " 当前无法确认继续接收边界");
+ Integer count = testMapper.selectCollectorChannelCount(battery.getCollectorId());
+ validateCollectorChannels(battery.getCollectorId(), count == null ? 0 : count,
+ battery.getCollectorChannels(), collectorChannels);
+ Date cabinetBoundary = Date.from(channel.getLastSeenAtUtc());
+ ensureAfterPreviousBoundary(current.getTestId(), "CABINET_CHANNEL", binding.getChannelId(), null,
+ cabinetBoundary);
+ intervals.add(interval(current, binding.getBindingId(), "CABINET_CHANNEL", binding.getChannelId(), null,
+ actionAt, cabinetBoundary, "继续接收：采用检测柜最近成功观测边界"));
+ for (TTestCollectorChannel mapping : mutable(battery.getCollectorChannels()))
+ {
+ Date collectorBoundary = collectorBoundary(battery.getCollectorId(), mapping.getChannelCode());
+ ensureAfterPreviousBoundary(current.getTestId(), "COLLECTOR_CHANNEL", battery.getCollectorId(),
+ mapping.getChannelCode(), collectorBoundary);
+ intervals.add(interval(current, binding.getBindingId(), "COLLECTOR_CHANNEL", battery.getCollectorId(),
+ mapping.getChannelCode(), actionAt, collectorBoundary, "继续接收：采用采集器最近成功观测边界"));
+ }
+ }
+ for (TTestReceptionInterval interval : intervals) testMapper.insertReceptionInterval(interval);
+ if (testMapper.updateReceptionResumed(testId, updateBy, actionAt) != 1)
+ throw new ServiceException("测试状态已变更，请刷新后重试");
+ testMapper.insertTestAudit(testId, "RECEPTION_RESUMED", operatorId, updateBy, null,
+ JSON.toJSONString(Map.of("receptionStatus", "RECEIVING", "intervalCount", intervals.size(),
+ "intervals", intervals)));
+ return 1;
+ }
+
+ @Override
+ public List<TTestReceptionInterval> selectReceptionIntervals(Long testId)
+ {
+ if (testId == null) throw new ServiceException("测试ID不能为空");
+ selectTestById(testId);
+ return mutable(testMapper.selectReceptionIntervals(testId));
+ }
+
+ private Date confirmBoundary(TTestReceptionInterval interval, Date actionAt)
+ {
+ if ("CABINET_CHANNEL".equals(interval.getSourceType()))
+ {
+ TDetectionChannel channel = testMapper.selectReceptionChannelForUpdate(interval.getSourceId());
+ if (channel == null || !isEnabled(channel.getStatus()) || !isOnline(channel.getOnlineStatus())
+ || channel.getLastSeenAtUtc() == null)
+ throw new ServiceException("检测柜通道 " + interval.getSourceId() + " 当前无法确认暂停接收边界");
+ Date boundary = Date.from(channel.getLastSeenAtUtc());
+ if (interval.getStartBoundaryAtUtc() != null && boundary.before(interval.getStartBoundaryAtUtc()))
+ throw new ServiceException("检测柜通道 " + interval.getSourceId() + " 的暂停边界早于区间开始");
+ return boundary;
+ }
+ if (!testMapper.collectorOnline(interval.getSourceId()))
+ throw new ServiceException("采集器 " + interval.getSourceId() + " 当前无法确认暂停接收边界");
+ Date boundary = collectorBoundary(interval.getSourceId(), interval.getSourceChannelCode());
+ if (interval.getStartBoundaryAtUtc() != null && boundary.before(interval.getStartBoundaryAtUtc()))
+ throw new ServiceException("采集器 " + interval.getSourceId() + " 的暂停边界早于区间开始");
+ return boundary;
+ }
+
+ private Date collectorBoundary(Long collectorId, String channelCode)
+ {
+ if (collectorSimulator == null) throw new ServiceException("采集器 " + collectorId + " 当前无法取得成功观测边界");
+ try
+ {
+ return collectorSimulator.collectorRealtime(collectorId.intValue()).measurements().stream()
+ .filter(item -> channelCode == null || channelCode.equals(item.channelCode()))
+ .map(CollectorRealtimeMeasurement::deviceTime).filter(java.util.Objects::nonNull)
+ .max(java.time.Instant::compareTo).map(Date::from)
+ .orElseThrow(() -> new ServiceException("采集器 " + collectorId + " 当前无法取得成功观测边界"));
+ }
+ catch (RuntimeException exception)
+ {
+ if (exception instanceof ServiceException serviceException) throw serviceException;
+ throw new ServiceException("采集器 " + collectorId + " 当前无法取得成功观测边界");
+ }
+ }
+
+ private void ensureAfterPreviousBoundary(Long testId, String sourceType, Long sourceId,
+ String sourceChannelCode, Date boundary)
+ {
+ TTestReceptionInterval previous = testMapper.selectLatestReceptionIntervalForUpdate(testId, sourceType,
+ sourceId, sourceChannelCode);
+ if (previous != null && previous.getEndBoundaryAtUtc() != null
+ && !boundary.after(previous.getEndBoundaryAtUtc()))
+ throw new ServiceException("数据来源尚未取得新的接收边界，请稍后重试");
+ }
+
+ private boolean isEndedWorkStatus(String workStatus)
+ {
+ return "STOPPED".equalsIgnoreCase(workStatus) || "FINISHED".equalsIgnoreCase(workStatus)
+ || "ENDED".equalsIgnoreCase(workStatus);
+ }
+
private TTestReceptionInterval interval(TTest test, Long bindingId, String sourceType, Long sourceId,
String sourceChannelCode, Date actionAt, Date boundaryAt, String note)
{
diff --git a/ruoyi-system/src/main/resources/mapper/system/TTestMapper.xml b/ruoyi-system/src/main/resources/mapper/system/TTestMapper.xml
index d432575..fd242ba 100644
--- a/ruoyi-system/src/main/resources/mapper/system/TTestMapper.xml
+++ b/ruoyi-system/src/main/resources/mapper/system/TTestMapper.xml
@@ -399,4 +399,52 @@
values(#{testId}, #{bindingId}, #{sourceType}, #{sourceId}, #{sourceChannelCode}, #{actionAtUtc},
#{startBoundaryAtUtc}, #{boundaryNote}, #{intervalStatus})
</insert>
+ <select id="selectOpenReceptionIntervalsForUpdate" parameterType="Long"
+ resultType="com.ruoyi.system.domain.TTestReceptionInterval">
+ select interval_id intervalId, test_id testId, binding_id bindingId, source_type sourceType,
+ source_id sourceId, source_channel_code sourceChannelCode, action_at_utc actionAtUtc,
+ start_boundary_at_utc startBoundaryAtUtc, end_boundary_at_utc endBoundaryAtUtc,
+ boundary_note boundaryNote, interval_status intervalStatus
+ from t_test_reception_interval
+ where test_id = #{testId} and interval_status = 'OPEN'
+ order by interval_id
+ for update
+ </select>
+ <select id="selectReceptionIntervals" parameterType="Long"
+ resultType="com.ruoyi.system.domain.TTestReceptionInterval">
+ select interval_id intervalId, test_id testId, binding_id bindingId, source_type sourceType,
+ source_id sourceId, source_channel_code sourceChannelCode, action_at_utc actionAtUtc,
+ start_boundary_at_utc startBoundaryAtUtc, end_boundary_at_utc endBoundaryAtUtc,
+ boundary_note boundaryNote, interval_status intervalStatus
+ from t_test_reception_interval
+ where test_id = #{testId}
+ order by interval_id
+ </select>
+ <select id="selectLatestReceptionIntervalForUpdate" resultType="com.ruoyi.system.domain.TTestReceptionInterval">
+ select interval_id intervalId, test_id testId, binding_id bindingId, source_type sourceType,
+ source_id sourceId, source_channel_code sourceChannelCode, action_at_utc actionAtUtc,
+ start_boundary_at_utc startBoundaryAtUtc, end_boundary_at_utc endBoundaryAtUtc,
+ boundary_note boundaryNote, interval_status intervalStatus
+ from t_test_reception_interval
+ where test_id = #{testId} and source_type = #{sourceType} and source_id = #{sourceId}
+ and (source_channel_code = #{sourceChannelCode} or (source_channel_code is null and #{sourceChannelCode} is null))
+ order by interval_id desc
+ limit 1
+ for update
+ </select>
+ <update id="closeReceptionInterval">
+ update t_test_reception_interval
+ set end_boundary_at_utc = #{endBoundaryAtUtc}, boundary_note = #{boundaryNote}, interval_status = 'CLOSED'
+ where interval_id = #{intervalId} and interval_status = 'OPEN'
+ </update>
+ <update id="updateReceptionPaused">
+ update t_test
+ set test_status = 'PAUSED', reception_status = 'PAUSED', update_by = #{updateBy}, update_time = #{updatedAtUtc}
+ where test_id = #{testId} and reception_status = 'RECEIVING' and del_flag = 0
+ </update>
+ <update id="updateReceptionResumed">
+ update t_test
+ set test_status = 'RUNNING', reception_status = 'RECEIVING', update_by = #{updateBy}, update_time = #{updatedAtUtc}
+ where test_id = #{testId} and reception_status = 'PAUSED' and del_flag = 0
+ </update>
</mapper>
--
Gitblit v1.7.1